stability improvements for multiprocessing, added queue for handling of return values

Author: konstantinpae

VERSION.txt

@@ -1 +1 @@
-14.4.5 experimental
+14.4.6 experimental

changelog.txt

@@ -18,3 +18,8 @@ Version 14.3.3 BETA
 Version 14.4.4 BETA
     - [minor] Updated module requests to ver 2.3.4 which also includes a new version of urllib3
 
+Version 14.4.6 experimental
+    - [major] added multiprocessing support
+    - [minor] some changes to the sensor's get_data function, especially the handling of the return values. Now handled by a queue.
+    - [minor] stability improvements to multiprocessing
+

probe.py

@@ -113,8 +113,9 @@ def main():
             if str(json_response) != '[]':
                 json_response_chunks = [json_response[i:i + 10] for i in range(0, len(json_response), 10)]
                 procs = []
-                out_queue = multiprocessing.Queue()
+
                 for element in json_response_chunks:
+                    out_queue = multiprocessing.Queue()
                     for part in element:
                         if config['debug']:
                             logging.debug(part)
@@ -130,10 +131,11 @@ def main():
                     for i in range(len(procs)):
                         json_payload_data.append(out_queue.get())
                     for p in procs:
-                        p.join()
+                        p.join(2)
                         procs = []
+                        p.terminate()
                         del p
-
+                    del out_queue
                     url_data = mini_probe.create_url(config, 'data')
                     try:
                         request_data = requests.post(url_data, data=json.dumps(json_payload_data), verify=False)
@@ -149,7 +151,7 @@ def main():
                         time.sleep((int(config['baseinterval']) * (9 / len(json_response))))
                     else:
                         time.sleep(int(config['baseinterval']) / 2)
-                del out_queue
+
             else:
                 logging.info("Nothing to do. Waiting for %s seconds." % (int(config['baseinterval']) / 3))
                 time.sleep(int(config['baseinterval']) / 3)

sensors/cpuload.py

@@ -78,7 +78,6 @@ def get_data(data, out_queue):
         del cpuload
         gc.collect()
         out_queue.put(data)
-        return data
 
     @staticmethod
     def read_cpu(path):

sensors/diskspace.py

@@ -77,7 +77,6 @@ def get_data(data, out_queue):
         del diskspace
         gc.collect()
         out_queue.put(data)
-        return data
 
     def read_disk(self):
         disks = []

sensors/http.py

@@ -205,4 +205,3 @@ def get_data(data, out_queue):
         del http
         gc.collect()
         out_queue.put(data)
-        return data

sensors/memory.py

@@ -78,7 +78,6 @@ def get_data(data, out_queue):
         del memory
         gc.collect()
         out_queue.put(data)
-        return data
 
     def read_memory(self, path):
         mem = open(path, "r")

sensors/ping.py

@@ -158,4 +158,3 @@ def get_data(data, out_queue):
         del ping
         gc.collect()
         out_queue.put(data)
-        return data

sensors/port.py

@@ -134,5 +134,4 @@ def get_data(data, out_queue):
         }
         del port
         gc.collect()
-        out_queue.put(data)
-        return data
+        out_queue.put(data)

sensors/probehealth.py

@@ -79,7 +79,6 @@ def get_data(data, out_queue):
             "channel": probedata
         }
         out_queue.put(data)
-        return data
 
     def read_memory(self, path):
         mem = open(path, "r")