API response logging, OrderedDict supported in stats, scheduler fixed - per-app and all-app tasks work now, sessions management fixed and less lock-demanding

Author: happz

src/api.py

@@ -52,27 +52,29 @@ def check_session():
 
   @staticmethod
   def run_handler():
-    hruntime.response.headers['Content-Type'] = 'application/json'
+    res = hruntime.response
+
+    res.headers['Content-Type'] = 'application/json'
 
     try:
       hlib.input.validate()
 
-      r = hruntime.request.handler(**hruntime.request.params)
+      res.api_response = hruntime.request.handler(**hruntime.request.params)
 
-      if r == None:
-        r = Reply(200)
+      if res.api_response == None:
+        res.api_response = Reply(200)
 
-      if hasattr(r, 'dump'):
-        return r.dump()
+      if hasattr(res.api_response, 'dump'):
+        return res.api_response.dump()
 
-      if hasattr(r, 'get_serializer'):
-        return r.get_serializer().serialize(r)
+      if hasattr(res.api_response, 'get_serializer'):
+        return res.api_response.get_serializer().serialize(res.api_response)
 
-      if type(r) == types.DictType:
-        return Raw(r).dump()
+      if type(res.api_response) == types.DictType:
+        return Raw(res.api_response).dump()
 
-      if type(r) in types.StringTypes:
-        return r
+      if type(res.api_response) in types.StringTypes:
+        return res.api_response
 
       raise hlib.error.InvalidOutputError()
 
@@ -86,8 +88,10 @@ def run_handler():
       hlib.log.log_error(e)
 
       kwargs = e.args_for_reply()
+      res.api_response = Reply(e.reply_status, error = Error(e), **kwargs)
+
       # pylint: disable-msg=W0142
-      return Reply(e.reply_status, error = Error(e), **kwargs).dump()
+      return res.api_response.dump()
 
   @staticmethod
   def redirect(url):

src/auth.py

@@ -46,6 +46,8 @@ def start_session(user = None, tainted = False):
   elif hasattr(hruntime.session, 'tainted'):
     hruntime.session.tainted = False
 
+  hruntime.app.sessions.invalidate_onlines()
+
   refresh_session()
 
 def check_session(redirect_to_login = True):
@@ -57,15 +59,15 @@ def check_session(redirect_to_login = True):
   """
 
   if not hruntime.request.is_authenticated:
+    hruntime.app.sessions.invalidate_onlines()
+
     if redirect_to_login == True:
       raise hlib.http.Redirect('/login/')
 
     return
 
   refresh_session()
 
-  hruntime.app.sessions.purge()
-
 def logout(trigger_event = True):
   """
   Mark session as not authenticated and redirect to login page.
@@ -77,4 +79,6 @@ def logout(trigger_event = True):
   hruntime.session.destroy()
   hruntime.session = None
 
+  hruntime.app.sessions.invalidate_onlines()
+
   raise hlib.http.Redirect('/login/')

src/cache.py

@@ -7,6 +7,8 @@
 import sys
 import threading
 
+from collections import OrderedDict
+
 import hlib.locks
 
 from hlib.stats import stats as STATS
@@ -23,16 +25,15 @@ def __init__(self, name, app):
 
     self.stats_name = 'Cache (%s - %s)' % (self.app.name, self.name)
 
-    with STATS:
-      STATS.set(self.stats_name, {
-        'Total objects': lambda s: sum([len(chain) for chain in self.objects.values()]),
-        'Total chains': lambda s: len(self.objects),
-        'Total size': lambda s: sum([sum([sys.getsizeof(v) for v in chain.values()]) for chain in self.objects.values()]),
-        'Hits': 0,
-        'Misses': 0,
-        'Inserts': 0,
-        'Chains': lambda s: self.to_stats()
-      })
+    STATS.set(self.stats_name, OrderedDict([
+      ('Total objects', lambda s: sum([len(chain) for chain in self.objects.values()])),
+      ('Total chains', lambda s: len(self.objects)),
+      ('Total size', lambda s: sum([sum([sys.getsizeof(v) for v in chain.values()]) for chain in self.objects.values()])),
+      ('Hits', 0),
+      ('Misses', 0),
+      ('Inserts', 0),
+      ('Chains', lambda s: self.to_stats())
+    ]))
 
   def stats_inc(self, key):
     with STATS:

src/database.py

@@ -13,6 +13,8 @@
 import persistent
 import threading
 
+from collections import OrderedDict
+
 import hlib.console
 import hlib.error
 import hlib.locks
@@ -143,16 +145,15 @@ def __init__(self, name, address, **kwargs):
     self.stats_name	= 'Database (%s)' % self.name
 
     # pylint: disable-msg=W0621
-    with STATS:
-      STATS.set(self.stats_name, {
-        'Loads': 0,
-        'Stores': 0,
-        'Commits': 0,
-        'Rollbacks': 0,
-        'Failed commits': 0,
-        'Connections': {},
-        'Caches': {}
-      })
+    STATS.set(self.stats_name, OrderedDict([
+      ('Loads', 0),
+      ('Stores', 0),
+      ('Commits', 0),
+      ('Rollbacks', 0),
+      ('Failed commits', 0),
+      ('Connections', {}),
+      ('Caches', {})
+    ]))
 
     import events
     events.Hook('engine.ThreadFinished', self.on_thread_finished)

src/engine.py

@@ -169,7 +169,7 @@ def log_access(self):
     self.__log(self.config['log.access.format'].format(**params), self.channels.access)
 
   def log_event(self, event):
-    self.__log('%s: %s' % (event.__class__.ename(), event.to_api()))
+    self.__log('%s: %s' % (event.__class__.ename(), event.to_api()), self.channels.events)
 
 class HeaderMap(UserDict.UserDict):
   """
@@ -419,6 +419,8 @@ def __init__(self):
     self.location		= None
     self.time			= hruntime.time
 
+    self.api_response = None
+
     self._output                = None
     self.output_length		= None
 
@@ -506,23 +508,27 @@ def dumps(self):
 
 class DataBaseGCTask(hlib.scheduler.Task):
   def __init__(self):
-    super(DataBaseGCTask, self).__init__('database-gc', self.database_gc, min = [5, 25, 45])
+    super(DataBaseGCTask, self).__init__('database-gc', self.database_gc, min = range(5, 60, 20))
 
   def database_gc(self, engine = None, app = None):
-    for app in engine.apps.values():
-      if app.db != None and app.db.is_opened:
-        app.db.globalGC()
+    if app.db != None and app.db.is_opened:
+      app.db.globalGC()
 
 class SaveSessionsTask(hlib.scheduler.Task):
   def __init__(self):
-    super(SaveSessionsTask, self).__init__('save-sessions', self.save_sessions, min = [10, 30, 50])
+    super(SaveSessionsTask, self).__init__('save-sessions', self.save_sessions, min = range(10, 60, 20))
 
   def save_sessions(self, engine = None, app = None):
-    print 'Save sessions task triggered'
+    if app.sessions != None:
+      app.sessions.save_sessions()
+
+class PurgeSessionsTask(hlib.scheduler.Task):
+  def __init__(self):
+    super(PurgeSessionsTask, self).__init__('purge-sessions', self.purge_sessions, min = range(0, 60, 5))
 
-    for app in engine.apps.values():
-      if app.sessions != None:
-        app.sessions.save_sessions()
+  def purge_sessions(self, engine = None, app = None):
+    if app.sessions != None:
+      app.sessions.purge()
 
 class Engine(object):
   """
@@ -545,6 +551,7 @@ def __init__(self, name, server_configs):
     self.scheduler.start()
 
     self.scheduler.add_task(DataBaseGCTask(), None)
+    self.scheduler.add_task(PurgeSessionsTask(), None)
     self.scheduler.add_task(SaveSessionsTask(), None)
 
     i = 0
@@ -661,14 +668,14 @@ def on_request_connected(self, _):
     with STATS:
       STATS.inc(self.stats_name, 'Total requests')
 
-    STATS.set(self.stats_name, 'Requests', hruntime.tid, {
-        'Bytes read':                     0,
-        'Bytes written':                  0,
-        'Client':                         hlib.server.ips_to_str(hruntime.request.ips),
-        'Start time':                     hruntime.time,
-        'Requested line':                 None,
-        'User':                           None
-      })
+    STATS.set(self.stats_name, 'Requests', hruntime.tid, OrderedDict([
+      ('Bytes read',     0),
+      ('Bytes written',  0),
+      ('Client',         hlib.server.ips_to_str(hruntime.request.ips)),
+      ('Start time',     hruntime.time),
+      ('Requested line', None),
+      ('User',           None)
+    ]))
 
   def on_request_accepted(self, _):
     STATS.set(self.stats_name, 'Requests', hruntime.tid, 'Client', hlib.server.ips_to_str(hruntime.request.ips))

src/error.py

@@ -13,6 +13,19 @@
 import traceback
 import types
 
+def get_caller(back = 0):
+  stack = traceback.extract_stack()
+
+  i = -3 - back
+  if abs(i) > len(stack):
+    i = -len(stack)
+
+  return stack[i]
+
+def fmt_caller(back = 0):
+  caller = get_caller(back = back + 1)
+  return 'Called by "%s" on %s:%i ("%s")' % (caller[2], caller[0], caller[1], caller[3])
+
 class BaseError(Exception):
   def __init__(self, msg = None, params = None, exception = None, exc_info = None, http_status = 500, reply_status = 500, dont_log = False, **kwargs):
     super(BaseError, self).__init__()
@@ -118,7 +131,7 @@ def __init__(self, obj = None):
     @param obj:			Unimplemented function wrapper that called this function.
     """
 
-    super(UnimplementedError, self).__init__('Unimplemented abstract method: %(method)s', params = {'method': UnimplementedError.function_name(obj, 2)})
+    super(UnimplementedError, self).__init__('Unimplemented abstract method: %(method)s', params = {'method': UnimplementedError.function_name(obj, 1)})
 
 class UnknownError(BaseError):
   pass

src/http/session.py

@@ -8,7 +8,9 @@
 import random
 import string
 import threading
-import UserDict
+
+from UserDict import UserDict
+from collections import OrderedDict
 
 import hlib.locks
 import hlib.server
@@ -23,9 +25,9 @@ def gen_rand_string(l):
   chars = string.letters + string.digits
   return reduce(lambda s, i: s + random.choice(chars), range(l), '')
 
-class Storage(UserDict.UserDict):
+class Storage(UserDict):
   def __init__(self, app, *args, **kwargs):
-    UserDict.UserDict.__init__(self, *args, **kwargs)
+    UserDict.__init__(self, *args, **kwargs)
 
     self.app = app
 
@@ -38,27 +40,39 @@ def __init__(self, app, *args, **kwargs):
 
     self.stats_name = 'Sessions (%s)' % self.app.name
 
-    STATS.set(self.stats_name, {
-      'Active': lambda s: ', '.join(self.online_users)
-    })
+    STATS.set(self.stats_name, OrderedDict([
+      ('Active', lambda s: ', '.join(self.online_users))
+    ]))
 
   @property
   def online_users(self):
+    # Some actions can be done without holding lock
+
+    __online_ref = self.__online
+    if __online_ref != None:
+      return __online_ref
+
+    if not self.sessions:
+      return []
+
     with self.lock:
-      if self.sessions == None:
-        return []
+      if self.__online != None:
+        return self.__online
 
-      if hruntime.time - self.__online_ctime > 60 or self.__online == None:
-        self.__online = []
+      self.__online = []
 
-        for session in self.sessions.values():
-          if session.age < 300 and hasattr(session, 'authenticated') and hasattr(session, 'name') and session.name:
-            self.__online.append(session.name)
+      for session in self.sessions.values():
+        if session.age < 300 and hasattr(session, 'authenticated') and hasattr(session, 'name') and session.name:
+          self.__online.append(session.name)
 
-        self.__online_ctime = hruntime.time
+      self.__online_ctime = hruntime.time
 
       return list(self.__online)
 
+  def invalidate_onlines(self):
+    with self.lock:
+      self.__online = None
+
   def purge(self):
     with self.lock:
       rm = []

src/log/__init__.py

@@ -40,6 +40,16 @@ def log_msg(msg, channels, flush = False):
 def log_params():
   import hlib.server
 
+  def __response_api_status():
+    res = hruntime.response
+
+    if not res.api_response:
+      return '-'
+    if not hasattr(res.api_response, 'status'):
+      return '/'
+
+    return res.api_response.status
+
   return {
     'tid':			hruntime.tid,
     'stamp':			hruntime.localtime,
@@ -51,6 +61,7 @@ def log_params():
     'request_agent':		hruntime.request.headers.get('User-Agent', '-'),
     'response_status':		hruntime.response.status,
     'response_length':		hruntime.response.output_length != None and hruntime.response.output_length or 0,
+    'response_api_status': __response_api_status(),
     'session_id':		hruntime.session.sid if hruntime.session != None else None
   }
 

src/log/channels/__init__.py

@@ -38,7 +38,8 @@ def __init__(self, stream, *args, **kwargs):
     self.stream		= stream
 
   def do_log_message(self, msg):
-    print >> self.stream, msg.encode('ascii', 'replace')
+    msg = msg.decode('utf-8', 'replace').encode('ascii', 'replace')
+    self.stream.write(msg + '\n')
     self.stream.flush()
 
   def log_error(self, error):