spacepaste

Paste details

reply | raw

  1.  
  2. import sys
  3. import zmq
  4. import time
  5. import logging
  6. from multiprocessing import Process
  7. import hashlib
  8. import binascii
  9. import struct
  11. logging.basicConfig(format="[%(asctime)s] " + \
  12. "[%(process)d %(threadName)s %(name)s] " + \
  13. "[%(module)s %(funcName)s %(lineno)d] " + \
  14. "[%(levelname)s] %(message)s", \
  15. level=logging.DEBUG)
  17. log = logging.getLogger(__name__)
  20. def hash(content):
  21. """
  22. Helper function, was being used
  23. to assist in ensuring durability
  24. but it was slowing things down.
  25. """
  26. return binascii.hexlify(
  27. hashlib.md5(
  28. pickle.dumps(content)
  29. ).digest()
  30. )
  32. put_url = 'ipc://msging.put'
  33. get_url = 'ipc://msging.get'
  34. shutdown_url = 'ipc://msging.shutdown'
  36. """
  37. Persistance layer, not in use because it slows things down.
  38. """
  39. pth = 'msging.persist'
  41. writer = open(pth, 'a')
  42. reader = open(pth, 'r+')
  44. def push(msg):
  45. data = pickle.dumps(msg)
  46. data = struct.pack('<L', len(data)) + data # size header
  47. data += '\x00' # marks if message has been read
  48. writer.write(data)
  50. def pop():
  51. while True:
  52. len, = struct.unpack('<L', reader.read(4))
  53. data = reader.read(len)
  54. if reader.read(1) != '\x00': # skipping read messages
  55. continue
  56. msg = pickle.loads(data)
  57. reader.seek(-1, 1)
  58. reader.write('\x01') # mark we have read message
  59. return msg
  62. def message_queue():
  64. log.debug("Creating context")
  65. context = zmq.Context()
  67. log.debug("Connecting to put socket")
  68. put = context.socket(zmq.PULL)
  69. put.bind(put_url)
  71. log.debug("Connecting to get socket")
  72. get = context.socket(zmq.REP)
  73. get.bind(get_url)
  75. log.debug("Connecting to shutdown socket")
  76. shutdown = context.socket(zmq.PULL)
  77. shutdown.bind(shutdown_url)
  79. log.debug("Connecting to poller")
  80. poller = zmq.Poller()
  81. poller.register(get, zmq.POLLIN)
  82. poller.register(put, zmq.POLLIN)
  83. poller.register(shutdown, zmq.POLLIN)
  85. lst = list()
  87. while True:
  89. comm = dict(poller.poll())
  91. if comm.get(put, False):
  92. msg = put.recv()
  93. lst.append(msg)
  96. if comm.get(get, False):
  97. get.recv()
  98. get.send(lst.pop())
  100. if comm.get(shutdown, False):
  101. log.warning("Shutting down")
  102. shutdown.recv()
  103. sys.exit()
  105. def put(msg, url = put_url):
  106. context = zmq.Context()
  107. put = context.socket(zmq.PUSH)
  108. put.connect(url)
  109. put.send(msg)
  111. def get(url = get_url):
  112. context = zmq.Context()
  113. get = context.socket(zmq.REQ)
  114. get.connect(url)
  115. get.send(str())
  116. msg = get.recv()
  117. return msg
  119. def shutdown(msg = None, url = shutdown_url):
  120. context = zmq.Context()
  121. shutdown = context.socket(zmq.PUSH)
  122. shutdown.connect(url)
  123. shutdown.send(str())
  125. mq = Process(target = message_queue)
  126. mq.start()
  127. time.sleep(1)
  129. max = 10000
  130. def produce():
  131. ts = time.time()
  132. count = 0
  133. try:
  134. for i in range(max):
  135. msg = "hello msg %d" % i
  136. put(msg)
  137. count += 1
  138. print "Sent %d msgs at rate of %s msg/sec." % (count, str(count / (time.time() - ts)))
  139. except:
  140. log.error(sys.exc_info())
  142. def retrieve():
  143. count = 0
  144. ts = time.time()
  145. try:
  146. for j in range(max):
  147. get()
  148. count += 1
  149. print "Recvd %d msgs at rate of %s msg/sec." % (count, str(count / (time.time() - ts)))
  150. except:
  151. log.error(sys.exc_info())
  153. p = Process(target = produce)
  154. p.start()
  155. time.sleep(0.1)
  157. r = Process(target = retrieve)
  158. r.start()
  160. p.join()
  161. r.join()
  163. shutdown()
  165. """
  166. Output:
  168. $ python test.py
  169. [2012-04-02 00:26:46,185] [31168 MainThread __main__] [test message_queue 71] [DEBUG] Creating context
  170. [2012-04-02 00:26:46,186] [31168 MainThread __main__] [test message_queue 74] [DEBUG] Connecting to put socket
  171. [2012-04-02 00:26:46,187] [31168 MainThread __main__] [test message_queue 78] [DEBUG] Connecting to get socket
  172. [2012-04-02 00:26:46,187] [31168 MainThread __main__] [test message_queue 82] [DEBUG] Connecting to shutdown socket
  173. [2012-04-02 00:26:46,187] [31168 MainThread __main__] [test message_queue 86] [DEBUG] Connecting to poller
  174. Sent 10000 msgs at rate of 1457.89044084 msg/sec.
  175. Recvd 10000 msgs at rate of 1244.68594864 msg/sec.
  176. [2012-04-02 00:26:55,330] [31168 MainThread __main__] [test message_queue 107] [WARNING] Shutting down
  177. """
  178.  
Powered by spacepaste. Paste removal.