<commit_before>'''
Created on Feb 25, 2012

@author: aw, cs
'''
import exceptions
import sys
import errno
import logging
import Queue
import socket

from pox.lib.util import assert_type, makePinger
from pox.lib.recoco import Select, Task
from pox.lib.ioworker.io_worker import IOWorker

log = logging.getLogger()

class DeferredIOWorker(object):
  '''
  Wrapper class for RecocoIOWorkers.

  Rather than actually sending/receiving right away, queue the messages.
  Then there are separate methods for actually sending the messages via
  the wrapped io_worker

  io_worker: io_worker to wrap
  call_later: recoco.Scheduler's call_later factory method for scheduling/thread-safety
  '''
  def __init__(self, io_worker, call_later):
    self._io_worker = io_worker
    self._io_worker.set_receive_handler(self.io_worker_receive_handler)
    self._call_later = call_later
    # Thread-safe read and write queues of indefinite length
    self._receive_queue = Queue.Queue()
    self._send_queue = Queue.Queue()
    # Read buffer that we present to clients
    self._receive_buf = ""

  def permit_send(self):
    '''
    deque()s the first element of the write queue, and actually sends it
    across the wire.

    raises an exception if the write queue is empty
    '''
    message = self._send_queue.get()
    # TODO(cs): will recoco guarentee in-order delivery of a sequence of these send requests?
    self._call_later(lambda: self._io_worker.send(message))

  def send(self, data):
    """ send data from the client side. fire and forget. """
    self._send_queue.put(data)

  def has_pending_sends(self):
    ''' called by the "arbitrator" in charge of deferal '''
    return not self._send_queue.empty()

  def permit_receive(self):
    '''
    deque()s the first element of the read queue, and notifies the client
    that there is data to be read.

    raises an exception if the read queue is empty
    '''
    data = self._receive_queue.get()
    self._receive_buf += data
    self._client_receive_handler(self)

  def set_receive_handler(self, block):
    ''' Called by client '''
    self._client_receive_handler = block

  def peek_receive_buf(self):
    ''' Called by client '''
    return self._receive_buf

  def consume_receive_buf(self, l):
    ''' called by client to consume receive buffer '''
    assert(len(self._receive_buf) >= l)
    self._receive_buf = self._receive_buf[l:]

  def io_worker_receive_handler(self, io_worker):
    ''' called from io_worker (recoco thread, after the Select loop pushes onto io_worker) '''
    # Consume everything immediately
    message = io_worker.peek_receive_buf()
    io_worker.consume_receive_buf(len(message))
    # thread-safe queue
    self._receive_queue.put(message)
<commit_msg>sts.deferred_io: s/message/data. deferred io only deals with bytes<commit_after>'''
Created on Feb 25, 2012

@author: aw, cs
'''
import exceptions
import sys
import errno
import logging
import Queue
import socket

from input_traces.fingerprints import OFFingerprint
from pox.lib.util import assert_type, makePinger
from pox.lib.recoco import Select, Task
from pox.lib.ioworker.io_worker import IOWorker

log = logging.getLogger()

class DeferredIOWorker(object):
  '''
  Wrapper class for RecocoIOWorkers.

  Rather than actually sending/receiving right away, queue the data.
  Then there are separate methods for actually sending the data via
  the wrapped io_worker

  io_worker: io_worker to wrap
  call_later: recoco.Scheduler's call_later factory method for scheduling/thread-safety
  '''
  def __init__(self, io_worker, call_later):
    self._io_worker = io_worker
    self._io_worker.set_receive_handler(self.io_worker_receive_handler)
    self._call_later = call_later
    # Thread-safe read and write queues of indefinite length
    self._receive_queue = Queue.Queue()
    self._send_queue = Queue.Queue()
    # Read buffer that we present to clients
    self._receive_buf = ""

  def permit_send(self):
    '''
    deque()s the first element of the write queue, and actually sends it
    across the wire.

    raises an exception if the write queue is empty
    '''
    data = self._send_queue.get()
    self._call_later(lambda: self._io_worker.send(data))

  def send(self, data):
    """ send data from the client side. fire and forget. """
    self._send_queue.put(data)

  def has_pending_sends(self):
    ''' called by the "arbitrator" in charge of deferal '''
    return not self._send_queue.empty()

  def permit_receive(self):
    '''
    deque()s the first element of the read queue, and notifies the client
    that there is data to be read.

    raises an exception if the read queue is empty
    '''
    data = self._receive_queue.get()
    self._receive_buf += data
    self._client_receive_handler(self)

  def set_receive_handler(self, block):
    ''' Called by client '''
    self._client_receive_handler = block

  def peek_receive_buf(self):
    ''' Called by client '''
    return self._receive_buf

  def consume_receive_buf(self, l):
    ''' called by client to consume receive buffer '''
    assert(len(self._receive_buf) >= l)
    self._receive_buf = self._receive_buf[l:]

  def io_worker_receive_handler(self, io_worker):
    ''' called from io_worker (recoco thread, after the Select loop pushes onto io_worker) '''
    # Consume everything immediately
    data = io_worker.peek_receive_buf()
    io_worker.consume_receive_buf(len(data))
    # thread-safe queue
    self._receive_queue.put(data)

{'fraction_non_alphanumeric': 0.08655953250547845, 'fraction_numerical': 0.002191380569758948, 'mean_word_length': 3.829805996472663, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2926843', 'n_tokens_mistral': 1737, 'n_tokens_neox': 1630, 'n_words': 524}