<commit_before> #!/usr/bin/env python

"""AMQP messaging with Pika."""

import gevent
from gevent import event, coros

from pika.credentials import PlainCredentials
from pika.connection import ConnectionParameters
from pika.adapters import SelectConnection
from pika import BasicProperties

from pyon.core.bootstrap import CFG
from pyon.net import amqp
from pyon.net import channel
from pyon.util.log import log

class IDPool(object):
    """
    Create a pool of IDs to allow reuse.
    The "new_id" function generates the next valid ID from the previous one. If not given, defaults to
    incrementing an integer.
    """

    def __init__(self, new_id=None):
        log.debug("In IDPool.__init__")
        if new_id is None: new_id = lambda x: x + 1
        log.debug("new_id: %s" % str(new_id))

        self.ids_in_use = set()
        self.ids_free = set()
        self.new_id = new_id
        self.last_id = 0

    def get_id(self):
        log.debug("In IDPool.get_id")
        log.debug("idsfree: %s", self.ids_free)
        log.debug("idsinuse: %s", self.ids_in_use)
        if len(self.ids_free) > 0:
            #log.debug("new_id: %s" % str(new_id))
            id = self.ids_free.pop()
            self.ids_in_use.add(id)
            log.debug("id: %s" % str(id))
            return id

        self.last_id = id_ = self.new_id(self.last_id)
        self.ids_in_use.add(id_)
        log.debug("id: %s" % str(id_))
        return id_

    def release_id(self, the_id):
        log.debug("In IDPool.release_id")
        log.debug("the_id: %s" % str(the_id))
        log.debug("idsfree: %s", self.ids_free)
        log.debug("idsinuse: %s", self.ids_in_use)
        if the_id in self.ids_in_use:
            self.ids_in_use.remove(the_id)
            self.ids_free.add(the_id)

class NodeB(amqp.Node):
    """
    Blocking interface to AMQP messaging primitives.

    Wrap around Node and create blocking interface for getting channel
    objects.
    """

    def __init__(self):
        log.debug("In NodeB.__init__")
        self.ready = event.Event()
        self._lock = coros.RLock()
        self._pool = IDPool()
        self._bidir_pool = {}   # maps inactive/active our numbers (from self._pool) to channels
        self._pool_map = {}     # maps active pika channel numbers to our numbers (from self._pool)

        amqp.Node.__init__(self)

    def start_node(self):
        """
        This should only be called by on_connection_opened.
        so, maybe we don't need a start_node/stop_node interface
        """
        log.debug("In start_node")
        amqp.Node.start_node(self)
        self.running = 1
        self.ready.set()

    def channel(self, ch_type):
        log.debug("In NodeB.channel, pool size is %d", len(self._bidir_pool))
        if not self.running:
            log.error("Attempt to open channel on node that is not running")
            raise #?

        log.debug("acquire semaphore")
        with self._lock:
            log.debug("acquired semaphore")

            def new_channel():
                result = event.AsyncResult()
                def on_channel_open_ok(amq_chan):
                    sch = ch_type(close_callback=self.on_channel_request_close)
                    sch.on_channel_open(amq_chan)
                    ch = channel.SocketInterface.Socket(amq_chan, sch)
                    result.set((ch, sch))
                self.client.channel(on_channel_open_ok)
                return result.get()

            if ch_type == channel.BidirectionalClient:
                chid = self._pool.get_id()
                if chid in self._bidir_pool:
                    assert not chid in self._pool_map
                    self._pool_map[self._bidir_pool[chid].amq_chan.channel_number] = chid
                    ch = self._bidir_pool[chid]
                    socket = channel.SocketInterface.Socket(ch.amq_chan, ch)
                else:
                    socket, ch = new_channel()
                    self._bidir_pool[chid] = ch
                    self._pool_map[ch.amq_chan.channel_number] = chid
            else:
                socket, ch = new_channel()

            assert socket and ch
            log.debug("sock channel: %s" % str(socket))

        log.debug("release semaphore")

        return socket

    def on_channel_request_close(self, ch):<commit_msg>Channel pooling bug - missed this when swap occured before<commit_after> #!/usr/bin/env python

"""AMQP messaging with Pika."""

import gevent
from gevent import event, coros

from pika.credentials import PlainCredentials
from pika.connection import ConnectionParameters
from pika.adapters import SelectConnection
from pika import BasicProperties

from pyon.core.bootstrap import CFG
from pyon.net import amqp
from pyon.net import channel
from pyon.util.log import log

class IDPool(object):
    """
    Create a pool of IDs to allow reuse.
    The "new_id" function generates the next valid ID from the previous one. If not given, defaults to
    incrementing an integer.
    """

    def __init__(self, new_id=None):
        log.debug("In IDPool.__init__")
        if new_id is None: new_id = lambda x: x + 1
        log.debug("new_id: %s" % str(new_id))

        self.ids_in_use = set()
        self.ids_free = set()
        self.new_id = new_id
        self.last_id = 0

    def get_id(self):
        log.debug("In IDPool.get_id")
        log.debug("idsfree: %s", self.ids_free)
        log.debug("idsinuse: %s", self.ids_in_use)
        if len(self.ids_free) > 0:
            #log.debug("new_id: %s" % str(new_id))
            id = self.ids_free.pop()
            self.ids_in_use.add(id)
            log.debug("id: %s" % str(id))
            return id

        self.last_id = id_ = self.new_id(self.last_id)
        self.ids_in_use.add(id_)
        log.debug("id: %s" % str(id_))
        return id_

    def release_id(self, the_id):
        log.debug("In IDPool.release_id")
        log.debug("the_id: %s" % str(the_id))
        log.debug("idsfree: %s", self.ids_free)
        log.debug("idsinuse: %s", self.ids_in_use)
        if the_id in self.ids_in_use:
            self.ids_in_use.remove(the_id)
            self.ids_free.add(the_id)

class NodeB(amqp.Node):
    """
    Blocking interface to AMQP messaging primitives.

    Wrap around Node and create blocking interface for getting channel
    objects.
    """

    def __init__(self):
        log.debug("In NodeB.__init__")
        self.ready = event.Event()
        self._lock = coros.RLock()
        self._pool = IDPool()
        self._bidir_pool = {}   # maps inactive/active our numbers (from self._pool) to channels
        self._pool_map = {}     # maps active pika channel numbers to our numbers (from self._pool)

        amqp.Node.__init__(self)

    def start_node(self):
        """
        This should only be called by on_connection_opened.
        so, maybe we don't need a start_node/stop_node interface
        """
        log.debug("In start_node")
        amqp.Node.start_node(self)
        self.running = 1
        self.ready.set()

    def channel(self, ch_type):
        log.debug("In NodeB.channel, pool size is %d", len(self._bidir_pool))
        if not self.running:
            log.error("Attempt to open channel on node that is not running")
            raise #?

        log.debug("acquire semaphore")
        with self._lock:
            log.debug("acquired semaphore")

            def new_channel():
                result = event.AsyncResult()
                def on_channel_open_ok(amq_chan):
                    sch = ch_type(close_callback=self.on_channel_request_close)
                    sch.on_channel_open(amq_chan)
                    ch = channel.SocketInterface.Socket(amq_chan, sch)
                    result.set((ch, sch))
                self.client.channel(on_channel_open_ok)
                return result.get()

            if ch_type == channel.BidirectionalClient:
                chid = self._pool.get_id()
                if chid in self._bidir_pool:
                    assert not chid in self._pool_map.values()
                    self._pool_map[self._bidir_pool[chid].amq_chan.channel_number] = chid
                    ch = self._bidir_pool[chid]
                    socket = channel.SocketInterface.Socket(ch.amq_chan, ch)
                else:
                    socket, ch = new_channel()
                    self._bidir_pool[chid] = ch
                    self._pool_map[ch.amq_chan.channel_number] = chid
            else:
                socket, ch = new_channel()

            assert socket and ch
            log.debug("sock channel: %s" % str(socket))

        log.debug("release semaphore")

        return socket

    def on_channel_request_close(self, ch):
{'fraction_non_alphanumeric': 0.10059578368469294, 'fraction_numerical': 0.0009165902841429881, 'mean_word_length': 2.030902777777778, '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': '1164950', 'n_tokens_mistral': 2749, 'n_tokens_neox': 2633, 'n_words': 651}