Python _dummy_thread 模块,allocate_lock() 实例源码

我们从Python开源项目中,提取了以下37个代码示例,用于说明如何使用_dummy_thread.allocate_lock()

项目:python-    作者:secondtonone1    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:deb-python-functools32    作者:openstack    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:ivaochdoc    作者:ivaoch    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:zippy    作者:securesystemslab    | 项目源码 | 文件源码
def setUp(self):
        # Create a lock
        self.lock = _thread.allocate_lock()
项目:zippy    作者:securesystemslab    | 项目源码 | 文件源码
def test_LockType(self):
        #Make sure _thread.LockType is the same type as _thread.allocate_locke()
        self.assertIsInstance(_thread.allocate_lock(), _thread.LockType,
                              "_thread.LockType is not an instance of what "
                              "is returned by _thread.allocate_lock()")
项目:zippy    作者:securesystemslab    | 项目源码 | 文件源码
def __init__(self, raw, buffer_size=DEFAULT_BUFFER_SIZE):
        """Create a new buffered reader using the given readable raw IO object.
        """
        if not raw.readable():
            raise IOError('"raw" argument must be readable.')

        _BufferedIOMixin.__init__(self, raw)
        if buffer_size <= 0:
            raise ValueError("invalid buffer size")
        self.buffer_size = buffer_size
        self._reset_read_buf()
        self._read_lock = Lock()
项目:zippy    作者:securesystemslab    | 项目源码 | 文件源码
def __init__(self, raw,
                 buffer_size=DEFAULT_BUFFER_SIZE, max_buffer_size=None):
        if not raw.writable():
            raise IOError('"raw" argument must be writable.')

        _BufferedIOMixin.__init__(self, raw)
        if buffer_size <= 0:
            raise ValueError("invalid buffer size")
        if max_buffer_size is not None:
            warnings.warn("max_buffer_size is deprecated", DeprecationWarning,
                          self._warning_stack_offset)
        self.buffer_size = buffer_size
        self._write_buf = bytearray()
        self._write_lock = Lock()
项目:zippy    作者:securesystemslab    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:news-for-good    作者:thecodinghub    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:yt    作者:yt-project    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:Tencent_Cartoon_Download    作者:Fretice    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:fieldsight-kobocat    作者:awemulya    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:web_ctp    作者:molebot    | 项目源码 | 文件源码
def setUp(self):
        # Create a lock
        self.lock = _thread.allocate_lock()
项目:web_ctp    作者:molebot    | 项目源码 | 文件源码
def test_LockType(self):
        #Make sure _thread.LockType is the same type as _thread.allocate_locke()
        self.assertIsInstance(_thread.allocate_lock(), _thread.LockType,
                              "_thread.LockType is not an instance of what "
                              "is returned by _thread.allocate_lock()")
项目:web_ctp    作者:molebot    | 项目源码 | 文件源码
def __init__(self, raw, buffer_size=DEFAULT_BUFFER_SIZE):
        """Create a new buffered reader using the given readable raw IO object.
        """
        if not raw.readable():
            raise IOError('"raw" argument must be readable.')

        _BufferedIOMixin.__init__(self, raw)
        if buffer_size <= 0:
            raise ValueError("invalid buffer size")
        self.buffer_size = buffer_size
        self._reset_read_buf()
        self._read_lock = Lock()
项目:web_ctp    作者:molebot    | 项目源码 | 文件源码
def __init__(self, raw, buffer_size=DEFAULT_BUFFER_SIZE):
        if not raw.writable():
            raise IOError('"raw" argument must be writable.')

        _BufferedIOMixin.__init__(self, raw)
        if buffer_size <= 0:
            raise ValueError("invalid buffer size")
        self.buffer_size = buffer_size
        self._write_buf = bytearray()
        self._write_lock = Lock()
项目:web_ctp    作者:molebot    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:splunk_ta_ps4_f1_2016    作者:jonathanvarley    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:TA-SyncKVStore    作者:georgestarcher    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:cb-defense-splunk-app    作者:carbonblack    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:CloudPrint    作者:William-An    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:ouroboros    作者:pybee    | 项目源码 | 文件源码
def setUp(self):
        # Create a lock
        self.lock = _thread.allocate_lock()
项目:ouroboros    作者:pybee    | 项目源码 | 文件源码
def test_LockType(self):
        #Make sure _thread.LockType is the same type as _thread.allocate_locke()
        self.assertIsInstance(_thread.allocate_lock(), _thread.LockType,
                              "_thread.LockType is not an instance of what "
                              "is returned by _thread.allocate_lock()")
项目:ouroboros    作者:pybee    | 项目源码 | 文件源码
def __init__(self, raw, buffer_size=DEFAULT_BUFFER_SIZE):
        """Create a new buffered reader using the given readable raw IO object.
        """
        if not raw.readable():
            raise OSError('"raw" argument must be readable.')

        _BufferedIOMixin.__init__(self, raw)
        if buffer_size <= 0:
            raise ValueError("invalid buffer size")
        self.buffer_size = buffer_size
        self._reset_read_buf()
        self._read_lock = Lock()
项目:ouroboros    作者:pybee    | 项目源码 | 文件源码
def __init__(self, raw, buffer_size=DEFAULT_BUFFER_SIZE):
        if not raw.writable():
            raise OSError('"raw" argument must be writable.')

        _BufferedIOMixin.__init__(self, raw)
        if buffer_size <= 0:
            raise ValueError("invalid buffer size")
        self.buffer_size = buffer_size
        self._write_buf = bytearray()
        self._write_lock = Lock()
项目:ouroboros    作者:pybee    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:gardenbot    作者:GoestaO    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:projeto    作者:BarmyPenguin    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:flask-zhenai-mongo-echarts    作者:Fretice    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:aweasome_learning    作者:Knight-ZXW    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:kbe_server    作者:xiaohaoppy    | 项目源码 | 文件源码
def setUp(self):
        # Create a lock
        self.lock = _thread.allocate_lock()
项目:kbe_server    作者:xiaohaoppy    | 项目源码 | 文件源码
def test_LockType(self):
        #Make sure _thread.LockType is the same type as _thread.allocate_locke()
        self.assertIsInstance(_thread.allocate_lock(), _thread.LockType,
                              "_thread.LockType is not an instance of what "
                              "is returned by _thread.allocate_lock()")
项目:kbe_server    作者:xiaohaoppy    | 项目源码 | 文件源码
def __init__(self, raw, buffer_size=DEFAULT_BUFFER_SIZE):
        """Create a new buffered reader using the given readable raw IO object.
        """
        if not raw.readable():
            raise OSError('"raw" argument must be readable.')

        _BufferedIOMixin.__init__(self, raw)
        if buffer_size <= 0:
            raise ValueError("invalid buffer size")
        self.buffer_size = buffer_size
        self._reset_read_buf()
        self._read_lock = Lock()
项目:kbe_server    作者:xiaohaoppy    | 项目源码 | 文件源码
def __init__(self, raw, buffer_size=DEFAULT_BUFFER_SIZE):
        if not raw.writable():
            raise OSError('"raw" argument must be writable.')

        _BufferedIOMixin.__init__(self, raw)
        if buffer_size <= 0:
            raise ValueError("invalid buffer size")
        self.buffer_size = buffer_size
        self._write_buf = bytearray()
        self._write_lock = Lock()
项目:kbe_server    作者:xiaohaoppy    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:blog_flask    作者:momantai    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()
项目:MyFriend-Rob    作者:lcheniv    | 项目源码 | 文件源码
def allocate_lock():
    """Dummy implementation of _thread.allocate_lock()."""
    return LockType()