|
|
| System | : | Linux MiraNet 3.0.0-14-generic-pae #23-Ubuntu SMP Mon Nov 21 22:07:10 UTC 2011 i686 |
|---|
| Software | : | Apache. PHP/5.3.6-13ubuntu3.10 |
|---|
| ID | : | uid=65534(nobody) gid=65534(nogroup) groups=65534(nogroup)
|
|---|
|
| Safe Mode | : | OFF |
|---|
| Open_Basedir | : | OFF |
|---|
| Freespace | : | 24.54 GB of 70.42 GB (34.85%) |
|---|
|
|
MySQL: ON MSSQL: OFF Oracle: OFF PostgreSQL: OFF Curl: OFF Sockets: ON Fetch: OFF Wget: ON Perl: ON |
|
Disabled Functions: pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,
|
[ System Info ]
[ Processes ]
[ SQL Manager ]
[ Eval ]
[ Encoder ]
[ Mailer ]
[ Back Connection ]
[ Backdoor Server ]
[ Kernel Exploit Search ]
[ MD5 Decrypter ]
[ Reverse IP ]
[ Kill Shell ]
[ FTP Brute-Force ]
|
|
/
usr/
share/
pyshared/
twisted/
test/
- drwxr-xr-x
|
Viewing file:  test_threadable.py (3.15 KB) -rw-r--r--Select action/file-type:  ( +) |  ( +) |  ( +) | Code ( +) | Session ( +) |  ( +) | SDB ( +) |  ( +) |  ( +) |  ( +) |  ( +) |  ( +) |
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
import sys, pickle
try:
import threading
except ImportError:
threading = None
from twisted.trial import unittest
from twisted.python import threadable
from twisted.internet import defer, reactor
class TestObject:
synchronized = ['aMethod']
x = -1
y = 1
def aMethod(self):
for i in xrange(10):
self.x, self.y = self.y, self.x
self.z = self.x + self.y
assert self.z == 0, "z == %d, not 0 as expected" % (self.z,)
threadable.synchronize(TestObject)
class SynchronizationTestCase(unittest.TestCase):
def setUp(self):
"""
Reduce the CPython check interval so that thread switches happen much
more often, hopefully exercising more possible race conditions. Also,
delay actual test startup until the reactor has been started.
"""
if hasattr(sys, 'getcheckinterval'):
self.addCleanup(sys.setcheckinterval, sys.getcheckinterval())
sys.setcheckinterval(7)
# XXX This is a trial hack. We need to make sure the reactor
# actually *starts* for isInIOThread() to have a meaningful result.
# Returning a Deferred here should force that to happen, if it has
# not happened already. In the future, this should not be
# necessary.
d = defer.Deferred()
reactor.callLater(0, d.callback, None)
return d
def testIsInIOThread(self):
foreignResult = []
t = threading.Thread(target=lambda: foreignResult.append(threadable.isInIOThread()))
t.start()
t.join()
self.failIf(foreignResult[0], "Non-IO thread reported as IO thread")
self.failUnless(threadable.isInIOThread(), "IO thread reported as not IO thread")
def testThreadedSynchronization(self):
o = TestObject()
errors = []
def callMethodLots():
try:
for i in xrange(1000):
o.aMethod()
except AssertionError, e:
errors.append(str(e))
threads = []
for x in range(5):
t = threading.Thread(target=callMethodLots)
threads.append(t)
t.start()
for t in threads:
t.join()
if errors:
raise unittest.FailTest(errors)
def testUnthreadedSynchronization(self):
o = TestObject()
for i in xrange(1000):
o.aMethod()
class SerializationTestCase(unittest.TestCase):
def testPickling(self):
lock = threadable.XLock()
lockType = type(lock)
lockPickle = pickle.dumps(lock)
newLock = pickle.loads(lockPickle)
self.failUnless(isinstance(newLock, lockType))
def testUnpickling(self):
lockPickle = 'ctwisted.python.threadable\nunpickle_lock\np0\n(tp1\nRp2\n.'
lock = pickle.loads(lockPickle)
newPickle = pickle.dumps(lock, 2)
newLock = pickle.loads(newPickle)
if threading is None:
SynchronizationTestCase.testThreadedSynchronization.skip = "Platform lacks thread support"
SerializationTestCase.testPickling.skip = "Platform lacks thread support"
|
