Thread's, async_chat and asyncore

Collapse
This topic is closed.
X
X
 
  • Time
  • Show
Clear All
new posts
  • Jos

    #1

    Thread's, async_chat and asyncore

    Hello.

    I'm using the asyncore and _chat modules to create a network server. I
    also have, running in a separate thread(s), a "producer" which needs to
    "push" data onto the network connection(s). (This is all for a
    multi-player game server, so the threads would be individual games,
    that need to update the players when new events happen in the game)

    I currently have a module level list of async_chat instances, which the
    thread then reads and uses the 'push' method of async_chat.

    I've done some very quick tests, but wanted to know more formally if
    the async_chat.push () method is thread safe?

    ie. can i be sure that the data that the thread pushes onto the
    connection will always be complete, and never "over-written" by another
    thread?

    Here is an example, which creates a whack of threads, which push their
    message down the first connection made to the server. I just use a
    telnet client to test it. I haven't seen any over-writing of values,
    but this is a very basic test/example of what i'm doing. I just want to
    make sure that this _is_ ok, or if i should be using a Queue...

    <file>
    import asynchat
    import asyncore
    import socket
    from threading import Thread
    import time
    import random

    clients = []

    class pulse_thread(Th read):
    def __init__(self, string):
    Thread.__init__ (self)
    self.msg = string

    def run(self):
    while True:
    time.sleep(rand om.random())
    if(len(clients) ):
    clients[0].push(self.msg)

    class Chat(asynchat.a sync_chat):

    def __init__(self, sock, addr, terminator="\r\ n", name="me"):

    asynchat.async_ chat.__init__(s elf, conn=sock)

    self.rmq = []
    self.terminator = terminator
    self.set_termin ator(self.termi nator)
    # push self onto module client list, so threads can access me
    clients.append( self)

    def collect_incomin g_data(self, data):
    """Buffer the data"""
    self.rmq.append (data)

    def found_terminato r(self):
    print "".join(self.rm q)
    self.rmq = []

    class cServer(asyncor e.dispatcher):

    # constructor method
    def __init__(self, addr, terminator="\r\ n"):

    # initalize asyncore
    asyncore.dispat cher.__init__(s elf)

    self.terminator = terminator

    # create a socket to listen on
    self.create_soc ket(socket.AF_I NET, socket.SOCK_STR EAM)

    # bind the socket to the ip/port
    self.bind(addr)

    # listen ??? don't know about the 5 ????
    self.listen(50)

    # EVENT handle_accept - fired when server accepts an new connection
    def handle_accept (self):
    # fire the accept method > returns a connection object
    conn, addr = self.accept()

    # extend the connection with cSock Class (another asyncore)
    #cSock(conn, addr)
    Chat(conn,addr, self.terminator )


    if __name__ == '__main__':

    for i in range(1,99):
    x = pulse_thread(": %02d:\n\r" % (i,))
    x.start()

    myserver = cServer(('127.0 .0.1',7000))

    asyncore.loop()
    </file>

    Please let me know if this is totally the _wrong_ way to do it!

    Thanks,
    Jos

  • Jos

    #2
    Re: Thread's, async_chat and asyncore

    Gaaa, i just did a run with 999 threads pushing data, and i'm getting
    "IndexError : deque" problems, so i'm guessing its _not_ thread safe...

    Bummer.

    jos

    Comment

    • Matt Hammond

      #3
      Re: Thread's, async_chat and asyncore

      Hi Jos,

      Have you looked at Kamaelia? Its a project I'm involved in to create a
      framework for highly concurrent systems, geared particularly for
      suitability for network server/client applications.

      A system is built out of many small components, each with their own
      pseudo-thread of execution (implemented as python generators). They
      communicate with each other by reading from and sending to local inboxes
      and outboxes. These are then linked together, causing messages to be
      passed from component to component.

      We're hopefully going to be making a new release soon, with some more
      interesting facilities. A Kamaelia system doing something similar could
      look something like this:

      ----code snippet starts----

      #!/usr/bin/env python

      from Axon.Component import component
      import time
      import random

      class PulseGenerator( component):
      def __init__(self, msg):
      super(PulseGene rator,self).__i nit__()
      self.msg = msg

      def main(self): # microprocess (generator)
      t=time.time()
      while 1:
      while time.time() < t:
      yield 1 # hand back control to scheduler
      t += random.random()
      self.send( self.msg, "outbox")


      if __name__ == "__main__":
      from Kamaelia.Util.S plitter import Plug, PlugSplitter
      from Kamaelia.Chassi s.ConnectedServ er import SimpleServer
      from Kamaelia.Util.p assThrough import passThrough
      from Axon.Scheduler import scheduler

      producer = PlugSplitter(Pu lseGenerator("h ello!\n")).acti vate()

      def newClientHandle r():
      return Plug(producer, passThrough()). activate()

      SimpleServer( protocol = newClientHandle r, port = 1601 ).activate()

      scheduler.run.r unThreads()

      ----code snippet ends----

      The SimpleServer component uses the factory function 'newClientHandl er' to
      create a component to handle the new client connection. In this case, its
      a 'plug' component that connects to a splitter component. The purpose of
      the splitter is to copy output to multiple destinations. In this case, the
      output comes from a PulseGenerator component.

      In the above system there's only one producer. To deal with multiple
      games, I guess you'd have to write a component that handles the initial
      'conversation' with the client to establish which game they want to
      connect to, before then creating the 'plug' component to connect to the
      appropriate game server.

      This is of course only one way to do it, and the 'plug' arrangement might
      not be the best eventual solution. We've also got a lookup-service, called
      the Co-ordinating Assistant Tracker. It translates name to references to
      inboxes or outboxes on components. You could, for example, use this
      service to look up components handling specific games.

      I'd encourage you to take a look at our project site on sourceforge
      ( http://kamaelia.sf.net/ ). There's some more introductory material and a
      tutorial or two which should give you a better idea about the system.

      We'd be interested to know what you think, and whether you think you could
      build your application using it.

      regards


      Matt


      --

      | Matt Hammond
      | R&D Engineer, BBC Research & Development, Tadworth, Surrey, UK.
      | http://kamaelia.sf.net/
      | http://www.bbc.co.uk/rd/

      Comment

      • Jos

        #4
        Re: Thread's, async_chat and asyncore

        I will take a look at it, thanks!
        jos

        Comment

        Working...