def close_all(self, only_receiving=True):
""" Close all channels in the token network.
Note: By default we're just discarding all channels we haven't received anything.
This potentially leaves deposits locked in channels after `closing`. This is "safe"
from an accounting point of view (deposits can not be lost), but may still be
undesirable from a liquidity point of view (deposits will only be freed after
manually closing or after the partner closed the channel).
If only_receiving is False then we close and settle all channels irrespective of them
having received transfers or not.
"""
with self.lock:
self.initial_channel_target = 0
channels_to_close = (
self.receiving_channels[:] if only_receiving else self.open_channels[:]
)
for channel in channels_to_close:
# FIXME: race condition, this can fail if channel was closed externally
self.api.close(self.token_address, channel.partner_address)
return channels_to_close
python类lock()的实例源码
def retry_connect(self):
"""Will be called when new channels in the token network are detected.
If the minimum number of channels was not yet established, it will try
to open new channels.
If the connection manager has no funds, this is a noop.
"""
# not initialized
if self.funds <= 0:
return
# in leaving state
if self.leaving_state:
return
with self.lock:
if self.funds_remaining <= 0:
return
if len(self.open_channels) >= self.initial_channel_target:
return
# try to fullfill our connection goal
self._add_new_partners()
def __init__( self, parent, socket ):
self.parent = parent
# A simple connection header sent by the proxy before the connection
# content, it encapsulates the original connection source information.
self.address = msgpack.unpackb( socket.recv( struct.unpack( '!I', socket.recv( 4 ) )[ 0 ] ) )
self.parent.log( 'Remote address: %s' % str( self.address ) )
try:
socket = parent.sslContext.wrap_socket( socket,
server_side = True,
do_handshake_on_connect = True,
suppress_ragged_eofs = True )
except:
raise DisconnectException
self.s = socket
self.aid = None
self.lock = Semaphore( 1 )
self.r = rpcm( isHumanReadable = True, isDebug = self.parent.log )
self.r.loadSymbols( Symbols.lookups )
self.connId = uuid.uuid4()
self.hostName = None
self.int_ip = None
self.ext_ip = None
self.tags = []
def _apply_rule(self, index, url, ret):
if ret.get('succ', False):
with lock:
self.succ_cnt += 1
def _apply_rule(self, index, url, ret):
if ret.get('succ', False):
with lock:
self.succ_cnt += 1
def __init__(self):
global _all_handles
# Generate label of text/unicode type from three random bytes.
self._id = codecs.encode(os.urandom(3), "hex_codec").decode("ascii")
self._legit_pid = os.getpid()
self._make_nonblocking()
self._lock = gevent.lock.Semaphore(value=1)
self._closed = False
_all_handles.append(self)
def __init__(self, api, token_address):
assert isinstance(api, RaidenAPI)
self.ready = Event()
self.api = api
self.token_address = token_address
existing_channels = self.api.get_channel_list(self.token_address)
open_channels = [
channel for channel in existing_channels if channel.state == CHANNEL_STATE_OPENED
]
if len(open_channels) == 0:
token = self.api.raiden.chain.token(self.token_address)
if not token.balance_of(self.api.raiden.address) > 0:
raise ValueError('not enough funds for echo node %s for token %s' % (
pex(self.api.raiden.address),
pex(self.token_address),
))
self.api.connect_token_network(
self.token_address,
token.balance_of(self.api.raiden.address),
initial_channel_target=10,
joinable_funds_target=.5,
)
self.last_poll_block = self.api.raiden.get_block_number()
self.received_transfers = Queue()
self.stop_signal = None # used to signal REMOVE_CALLBACK and stop echo_workers
self.greenlets = list()
self.lock = BoundedSemaphore()
self.seen_transfers = deque(list(), TRANSFER_MEMORY)
self.num_handled_transfers = 0
self.lottery_pool = Queue()
# register ourselves with the raiden alarm task
self.api.raiden.alarm.register_callback(self.echo_node_alarm_callback)
self.echo_worker_greenlet = gevent.spawn(self.echo_worker)
def __init__(
self,
raiden,
token_address,
channelgraph):
self.lock = Semaphore()
self.raiden = raiden
self.api = RaidenAPI(raiden)
self.channelgraph = channelgraph
self.token_address = token_address
self.funds = 0
self.initial_channel_target = 0
self.joinable_funds_target = 0
def _sendmsg(self, name, arg=None, uuid="", lock=False, loops=1,
def _protocol_sendmsg(self, name, args=None, uuid="", lock=False, loops=1,
def __init__( self, parent, socket ):
self.parent = parent
# A simple connection header sent by the proxy before the connection
# content, it encapsulates the original connection source information.
self.address = msgpack.unpackb( socket.recv( struct.unpack( '!I', socket.recv( 4 ) )[ 0 ] ) )
self.parent.log( 'Remote address: %s' % str( self.address ) )
try:
socket = parent.sslContext.wrap_socket( socket,
server_side = True,
do_handshake_on_connect = True,
suppress_ragged_eofs = True )
except:
raise DisconnectException
self.s = socket
self.aid = None
self.lock = Semaphore( 1 )
self.r = rpcm( isHumanReadable = True, isDebug = self.parent.log )
self.r.loadSymbols( Symbols.lookups )
self.connId = uuid.uuid4()
self.hostName = None
self.int_ip = None
self.ext_ip = None
self.tags = []
self.n_frames_received = 0
def close( self ):
with self.lock:
self.s.close()
def sendData( self, data, timeout = None ):
timeout = gevent.Timeout( timeout )
timeout.start()
try:
with self.lock:
self.s.sendall( data )
except:
raise DisconnectException( 'disconnect while sending' )
finally:
timeout.cancel()
def close( self ):
with self.lock:
self.s.close()
def __init__( self ):
self._sem = gevent.lock.BoundedSemaphore( value = 1 )
def lock( self, timeout = None ):
return self._sem.acquire( blocking = True, timeout = timeout )
def __enter__( self ):
self.lock()
def __init__( self, nReaders ):
self._nReaders = nReaders
self._sem = gevent.lock.BoundedSemaphore( value = nReaders )
def synchronized( f ):
'''Synchronization decorator.'''
lock = Mutex()
def new_function( *args, **kw ):
lock.lock()
try:
return f( *args, **kw )
finally:
lock.unlock()
return new_function
def count_down(self):
with self._lock:
assert self._count > 0
self._count -= 1
# Return inside lock to return the correct value,
# otherwise another thread could already have
# decremented again.
return self._count
def _execute(self):
"""
Executes the Eventual function, guarded by a lock.
"""
with self._mutex:
if self._waiter_greenlet:
self._waiter_greenlet.kill()
self._waiter_greenlet = None
self.function()
self._next_execution_time = None
def load(self, ctx):
super(InternalPlugin, self).load(ctx)
self.events = RedisSet(rdb, 'internal:tracked-events')
self.session_id = None
self.lock = Semaphore()
self.cache = []
def on_gateway_event(self, event):
if event['t'] not in self.events:
return
with self.lock:
self.cache.append(event)
def flush_cache(self):
while True:
gevent.sleep(1)
if not len(self.cache):
continue
with self.lock:
Event.insert_many(filter(bool, [
Event.prepare(self.session_id, event) for event in self.cache
])).execute()
self.cache = []
def __init__(self, name, task):
self.name = name
self.task = task
self.lock = Semaphore(task.max_concurrent)
def poll_all_received_events(self):
""" This will be triggered once for each `echo_node_alarm_callback`.
It polls all channels for `EventTransferReceivedSuccess` events,
adds all new events to the `self.received_transfers` queue and
respawns `self.echo_node_worker`, if it died. """
locked = False
try:
with Timeout(10):
locked = self.lock.acquire(blocking=False)
if not locked:
return
else:
channels = self.api.get_channel_list(token_address=self.token_address)
received_transfers = list()
for channel in channels:
channel_events = self.api.get_channel_events(
channel.channel_address,
self.last_poll_block
)
received_transfers.extend([
event for event in channel_events
if event['_event_type'] == 'EventTransferReceivedSuccess'
])
for event in received_transfers:
transfer = event.copy()
transfer.pop('block_number')
self.received_transfers.put(transfer)
# set last_poll_block after events are enqueued (timeout safe)
if received_transfers:
self.last_poll_block = max(
event['block_number']
for event in received_transfers
)
# increase last_poll_block if the blockchain proceeded
delta_blocks = self.api.raiden.get_block_number() - self.last_poll_block
if delta_blocks > 1:
self.last_poll_block += 1
if not self.echo_worker_greenlet.started:
log.debug(
'restarting echo_worker_greenlet',
dead=self.echo_worker_greenlet.dead,
successful=self.echo_worker_greenlet.successful(),
exception=self.echo_worker_greenlet.exception
)
self.echo_worker_greenlet = gevent.spawn(self.echo_worker)
except Timeout:
log.info('timeout while polling for events')
finally:
if locked:
self.lock.release()
def connect(
self,
funds,
initial_channel_target=3,
joinable_funds_target=.4):
"""Connect to the network.
Use this to establish a connection with the token network.
Subsequent calls to `connect` are allowed, but will only affect the spendable
funds and the connection strategy parameters for the future. `connect` will not
close any channels.
Note: the ConnectionManager does not discriminate manually opened channels from
automatically opened ones. If the user manually opened channels, those deposit
amounts will affect the funding per channel and the number of new channels opened.
Args:
funds (int): the amount of tokens spendable for this
ConnectionManager.
initial_channel_target (int): number of channels to open immediately
joinable_funds_target (float): amount of funds not initially assigned
"""
if funds <= 0:
raise ValueError('connecting needs a positive value for `funds`')
if self.token_address in self.raiden.message_handler.blocked_tokens:
self.raiden.message_handler.blocked_tokens.pop(self.token_address)
self.initial_channel_target = initial_channel_target
self.joinable_funds_target = joinable_funds_target
open_channels = self.open_channels
# there are already channels open
if len(open_channels):
log.debug(
'connect() called on an already joined token network',
token_address=pex(self.token_address),
open_channels=len(open_channels),
sum_deposits=self.sum_deposits,
funds=funds,
)
if len(self.channelgraph.graph.nodes()) == 0:
with self.lock:
log.debug('bootstrapping token network.')
# make ourselves visible
self.api.open(
self.token_address,
ConnectionManager.BOOTSTRAP_ADDR
)
with self.lock:
# set our available funds
self.funds = funds
# try to fullfill our connection goal
self._add_new_partners()
