[Nym3-commit] r517 - trunk/nymbaron/Server
jr at conuropsis.org
jr at conuropsis.org
Tue Jun 20 00:47:09 CEST 2006
Author: jr
Date: 2006-06-20 00:47:09 +0200 (Tue, 20 Jun 2006)
New Revision: 517
Modified:
trunk/nymbaron/Server/User.py
Log:
- start using the server send queue
Modified: trunk/nymbaron/Server/User.py
===================================================================
--- trunk/nymbaron/Server/User.py 2006-06-11 16:14:48 UTC (rev 516)
+++ trunk/nymbaron/Server/User.py 2006-06-19 22:47:09 UTC (rev 517)
@@ -34,6 +34,7 @@
import pickle
import time
import calendar
+import logging
import mixminion.Common
import mixminion.Crypto as _cr
@@ -434,55 +435,137 @@
"""
return self.config.path + os.sep + self.data['username'] + '.surbs'
- def advanced_send(self, msg, add_status = True, fun = None, funargs = None):
+ def advanced_send(self, msg, fun = None, funargs = None,
+ add_status = True, enqueue_only = False):
"""Schedules the sending of a message to the nymholder through the
mixminion network, using the surbs provided by the nymholder, and try to
send the elements present in the send queue. fun is a method of the User
object which will be called with argument funargs upon successful
sending of the message.
+ If enqueue_only is True, don't try to send anything, just queue msg.
#TODO add_status will be removed in the near future.
Adds a status control message if the add_status argument is True. Also
- updates the number of surbs"""
- self._load_queue()
- #TODO add surb usage management
- #TODO add the filling of the remaining place available in the the
- #message by syns and msgs here, cf 2.3 items 3
+ updates the number of surbs
+ fo the time being returns the result of send, or 2 if send was not
+ called. Will be removed when Main uses the success function mechanism"""
+ logger = logging.getLogger('nymbaron')
+ self.load_queue()
+ logger.debug('queue length (after loading): %d' % len(self.queue))
+ self.queue.append((msg, fun, funargs))
+ logger.debug('queue length (after appending): %d' % len(self.queue))
+ if enqueue_only:
+ return 2
+ #TODO add surb usage management : how many surb can we use?
+
+ #how many control message can we send?
+ #after the loop sMsg contains the concatenation of messages that will be
+ #sent, i the index in the queue of the first message not sent (the size
+ #of the queue if everything is sent
+ sMsg = ""
+ i = 0
+ nb_surb_max = 0
+ for (msg, fun, funargs) in self.queue:
+ tMsg = sMsg
+ tnb = nb_surb_max
+ sMsg = sMsg + msg
+ nb_surb_max = getNPacketsToEncode(sMsg, 0)
+ if nb_surb_max > self['nSurbs']:
+ sMsg = tMsg
+ nb_surb_max = tnb
+ break
+ i = i + 1
+ #if there are not enough surbs to send anything, we leave here
+ if i == 0:
+ logger.debug('Not enough surbs (%d) to send any control message' %
+ self['nSurbs'])
+ return 2
+ logger.debug('trying to send %d control messages' % i)
+ #else
+ #nb_surb_max is the number of surbs needed to send the control messages
+ #we can send
+
+ #compute a status message. It will be sent if there are enoughe surbs
+ #left
statusc = Message.Status()
nMsg = 0
usage = 0
- cmsg = msg
- if add_status:
- #determine the status elements, the number of surbs is assumed
- #to stay equal to its value
- self.load_index()
- for _, v in self.index.iteritems():
- if v['status'] in [lifeCycle['nothing-sent'], lifeCycle['synopsis-sent']]:
- nMsg += 1
- usage = self.usage()
- statusc.fromData(nMsg, 0, self.quota(), usage, self['received'])
- cmsg += str(statusc)
- nb_req_surb = getNPacketsToEncode(cmsg, 0)
- if add_status:
- while True:
- statusc.fromData(nMsg, self['nSurbs'] - nb_req_surb,
- self.quota(), usage, self['received'])
- cmsg = msg + str(statusc)
- act_nb = getNPacketsToEncode(cmsg, 0)
- old_nb_req_surb = nb_req_surb
- nb_req_surb = act_nb
- if nb_req_surb <= old_nb_req_surb:
- break
- if nb_req_surb > self['nSurbs']:
- #TODO
- # - send an Error message "low on surbs" if a surb is available
- # - store the message we couldn't send for later
- return 1
- #TODO debbuging cruft
- print "%d surbs used to send control messages" % nb_req_surb
+ #determine the status elements, the number of surbs is assumed
+ #to stay equal to its value
+ self.load_index()
+ for _, v in self.index.iteritems():
+ if v['status'] in \
+ [lifeCycle['nothing-sent'], lifeCycle['synopsis-sent']]:
+ nMsg += 1
+ usage = self.usage()
+ statusc.fromData(nMsg, 0, self.quota(), usage, self['received'])
+ scMsg = str(statusc)
+
+ #decide whether we add a status control message
+ nb_req_surb = getNPacketsToEncode(sMsg + scMsg, 0)
+ if nb_req_surb <= nb_surb_max:
+ cMsg = sMsg + scMsg
+ send_status = True
+ logger.debug('1st pass : OK to add status')
+ else:
+ cMsg = sMsg
+ nb_req_surb = nb_surb_max
+ send_status = False
+ logger.debug('1st pass : NOK to add status')
+ #TODO add the filling of the remaining place available in the the
+ #message by syns and msgs here, cf 2.3 items 3
+
+ #finalize the number of surbs to use
+ if send_status:
+ statusc.fromData(nMsg, self['nSurbs'] - nb_req_surb,
+ self.quota(), usage, self['received'])
+ cMsg = sMsg + str(statusc)
+ act_nb = getNPacketsToEncode(cMsg, 0)
+ old_nb_req_surb = nb_req_surb
+ nb_req_surb = act_nb
+ if nb_req_surb > old_nb_req_surb:
+ logger.debug('2st pass : NOK to add status')
+ send_status = False
+ cMsg = sMsg
+ nb_req_surb = nb_surb_max
+
+ #old logic, might still be useful in upcoming evolution
+ #while True:
+ # statusc.fromData(nMsg, self['nSurbs'] - nb_req_surb,
+ # self.quota(), usage, self['received'])
+ # cmsg = msg + str(statusc)
+ # act_nb = getNPacketsToEncode(cmsg, 0)
+ # old_nb_req_surb = nb_req_surb
+ # nb_req_surb = act_nb
+ # if nb_req_surb <= old_nb_req_surb:
+ # break
+
+ logger.debug('%d surbs used to send control messages' % nb_req_surb)
+
+ #actually do the sending, we assume here than whatever happens the surbs
+ #will be used TODO call clean surbs if an error happens
self['nSurbs'] -= nb_req_surb
- self['received'] = []
- return self.send(cmsg)
-
+ ec = self.send(cMsg)
+
+ #if the sending was successful
+ if ec == 0:
+ logger.debug('control messages successfully sent')
+ if send_status:
+ #TODO maybe we can implement a finer strategy, to reduce the
+ #number of surbs used by a status ctrl message
+ self['received'] = []
+ #clean the queue
+ for j in range(i):
+ msg, fun, funargs = self.queue[j]
+ if fun != None:
+ if funargs != None:
+ fun(*funargs)
+ else:
+ fun()
+ self.queue = self.queue[i:]
+ logger.debug('queue length (after sending): %d' % len(self.queue))
+
+ #ouf
+ return ec
def send(self, msg):
"""Sends a message to the nymholder through the mixminion network,
More information about the Nym3-commit
mailing list