Added threads parameter to server (Issue #28)

jip/cli/jip_server.py

@@ -15,16 +15,22 @@
     }
 
 Usage:
-   jip-server--help|-h] [-p <port>]
+   jip-server--help|-h] [-p <port>] [-t <threads>] [-l <level>]
 
 Options:
-    -p, --port <port >  The port used for the server
-                        [default: 5556]
+    -p, --port <port >       The port used for the server
+                             [default: 5556]
+    -t, --threads <threads>  Number of available parallel slots.
+                             Defaults to the number of cpu's detected
+                             in the system.
+    -l, --loglevel <level>   The log level. On of DEBUG, INFO, WARN, ERROR
+                             [default: INFO]
 
 Other Options:
     -h --help             Show this help message
 """
 
+import logging
 
 from jip.logger import getLogger
 from . import parse_args
@@ -53,16 +59,27 @@ def main():
 
     try:
         port = args['--port']
-        log.info("Starting JIP grid server no port %s", port)
+        threads = args['--threads']
+        threads = threads if threads is None else int(threads)
+        level = args['--loglevel']
+        level = {"INFO": logging.INFO,
+                 "WARN": logging.WARN,
+                 "DEBUG": logging.DEBUG,
+                 "ERROR": logging.ERROR}.get(level.upper(), None)
+        if level is None:
+            print >>sys.stderr, "Unknown log level:", args['--level']
+            sys.exit(1)
+        log.setLevel(level)
+        log.info("Starting JIP grid server on port %s", port)
         context = zmq.Context()
         socket = context.socket(zmq.REP)
         socket.bind("tcp://*:%s" % port)
         log.info("Socket server started")
 
-        cluster = jip.grids.LocalCluster()
+        cluster = jip.grids.LocalCluster(cores=threads)
         while True:
             msg = socket.recv_json()
-            log.info("Received message: %s" % msg)
+            log.debug("Received message: %s" % msg)
             if msg['cmd'] == 'list':
                 jobs = cluster.list()
                 log.debug("Recevied cluster jobs: %s", jobs)

jip/grids.py

@@ -6,21 +6,28 @@
 import subprocess
 import signal
 import sys
+import logging
 
 import jip.cluster
 from jip.logger import getLogger
 
 
 class LocalCluster(jip.cluster.Cluster):
 
-    def __init__(self, _start=True, _remote_ids=False):
+    def __init__(self, _start=True,
+                 _remote_ids=False,
+                 cores=None, log_level=logging.INFO):
         self._current_id = 0
         self.log = getLogger("jip.grids.LocalCluster")
         self.master_requests = None
         self.master_response = None
         self.master_process = None
         self._current_id
         self._remote_ids = _remote_ids
+        self.cores = cores
+        self.loglevel = log_level
+        # set log level
+        self.log.setLevel(log_level)
         # start the mater process
         if _start:
             self.start()
@@ -47,6 +54,8 @@ def start(self):
             self.master_process = multiprocessing.Process(
                 target=_GridMaster.create_master,
                 args=[self.master_requests, self.master_response],
+                kwargs={"cores": self.cores,
+                        "loglevel": self.loglevel},
                 name="grid-master"
             )
             self.master_process.start()
@@ -250,7 +259,7 @@ def __repr__(self):
 
 class _GridMaster(object):
     """The grid master instance"""
-    def __init__(self, requests, response, cores=None):
+    def __init__(self, requests, response, cores=None, loglevel=logging.INFO):
         """Initialize a new grid master with the request and response
         queues and optionally the number of cores or slots available.
         If no cores are specified, the number of cores of the current
@@ -276,7 +285,8 @@ def __init__(self, requests, response, cores=None):
         #: current id
         self._current_id = 0
 
-        self.log.info("Master | Initialized with %s available slots",
+        self.log.setLevel(loglevel)
+        self.log.info("Master | Initialized with %d available slots",
                       self.slots_available)
 
     def _next_id(self):
@@ -549,8 +559,9 @@ def _num_jobs(self):
         return len(self.queued) + len(self.running)
 
     @staticmethod
-    def create_master(request, response):
-        master = _GridMaster(request, response)
+    def create_master(request, response, cores=None, loglevel=logging.INFO):
+        master = _GridMaster(request, response, cores=cores,
+                             loglevel=loglevel)
         master.start()