Merge pull request #5 from tqsd/process_pool

benjione · web-flow · commit a7ecbfc21d93 · 2020-03-23T16:28:37.000+01:00
Process pool
diff --git a/eqsn/gates.py b/eqsn/gates.py
@@ -3,8 +3,10 @@
 import numpy as np
 from eqsn.qubit_thread import SINGLE_GATE, MERGE_SEND, MERGE_ACCEPT, MEASURE,\
                 MEASURE_NON_DESTRUCTIVE, GIVE_QUBITS_AND_TERMINATE, \
-                CONTROLLED_GATE, QubitThread
+                CONTROLLED_GATE, NEW_QUBIT, ADD_MERGED_QUBITS_TO_DICT
 from eqsn.shared_dict import SharedDict
+from eqsn.worker_process import WorkerProcess
+from eqsn.process_picker import ProcessPicker
 
 
 class EQSN(object):
@@ -17,19 +19,28 @@ class EQSN(object):
     def __init__(self):
         self.manager = multiprocessing.Manager()
         self.shared_dict = SharedDict.get_instance()
+        # cpu_count = multiprocessing.cpu_count()
+        cpu_count = 1
+        process_queue_list = []
+        for _ in range(cpu_count):
+            q = multiprocessing.Queue()
+            new_worker = WorkerProcess(q)
+            p = multiprocessing.Process(target=new_worker.run, args=())
+            p.start()
+            process_queue_list.append((p, q))
+        self.process_picker = ProcessPicker.get_instance(
+            cpu_count, process_queue_list)
 
     def new_qubit(self, q_id):
         """
         Creates a new qubit with an id.
 
         Args:
-            id (String): Id of the new qubit.
+            q_id (String): Id of the new qubit.
         """
-        q = multiprocessing.Queue()
-        thread = QubitThread(q_id, q)
-        p = multiprocessing.Process(target=thread.run, args=())
+        p, q = self.process_picker.get_next_process_queue()
+        q.put([NEW_QUBIT, q_id])
         self.shared_dict.set_thread_with_id(q_id, p, q)
-        p.start()
         logging.debug("Created new qubit with id %s.", q_id)
 
     def stop_all(self):
@@ -38,6 +49,8 @@ def stop_all(self):
         """
         self.shared_dict.send_all_threads(None)
         self.shared_dict.stop_all_threads()
+        self.shared_dict.stop_shared_dict()
+        self.process_picker.stop_process_picker()
 
     def X_gate(self, q_id):
         """
@@ -144,12 +157,13 @@ def merge_qubits(self, q_id1, q_id2):
             q1 = l[0]
             q2 = l[1]
             merge_q = self.manager.Queue()
-            q1.put([MERGE_SEND, merge_q])
-            q2.put([MERGE_ACCEPT, merge_q])
+            q1.put([MERGE_SEND, q_id1, merge_q])
+            q2.put([MERGE_ACCEPT, q_id2, merge_q])
             qubits_q = self.manager.Queue()
-            q1.put([GIVE_QUBITS_AND_TERMINATE, qubits_q])
+            q1.put([GIVE_QUBITS_AND_TERMINATE, q_id1, qubits_q])
             qubits = qubits_q.get()
-            self.shared_dict.change_thread_and_queue_of_ids_and_join(
+            q2.put([ADD_MERGED_QUBITS_TO_DICT, q_id2, qubits])
+            self.shared_dict.change_thread_and_queue_of_ids(
                 qubits, q_id2)
 
     def cnot_gate(self, q_id1, q_id2):
diff --git a/eqsn/process_picker.py b/eqsn/process_picker.py
@@ -0,0 +1,32 @@
+class ProcessPicker(object):
+    """
+    Decides which process allocates new Qubits.
+    """
+    __instance = None
+
+    @staticmethod
+    def get_instance(cpu_count, process_queue_list):
+        if ProcessPicker.__instance is None:
+            return ProcessPicker(cpu_count, process_queue_list)
+        return ProcessPicker.__instance
+
+    def __init__(self, cpu_count, process_queue_list):
+        if ProcessPicker.__instance is not None:
+            raise ValueError(
+                "Use get instance to get the Process picker class.")
+        ProcessPicker.__instance = self
+        self.amount_processes = cpu_count
+        self.pointer = 0
+        self.process_queue_list = process_queue_list
+
+    def get_next_process_queue(self):
+        """
+        Called in threads may override pointer, but we do not care, since
+        the pointer value is not essential.
+        """
+        res_q = self.process_queue_list[self.pointer % self.amount_processes]
+        self.pointer += 1
+        return res_q
+
+    def stop_process_picker(self):
+        ProcessPicker.__instance = None
diff --git a/eqsn/qubit_thread.py b/eqsn/qubit_thread.py
@@ -12,6 +12,8 @@
 MERGE_SEND = 5
 GIVE_QUBITS_AND_TERMINATE = 6
 MEASURE_NON_DESTRUCTIVE = 7
+NEW_QUBIT = 8
+ADD_MERGED_QUBITS_TO_DICT = 9
 
 
 class QubitThread(object):
@@ -212,18 +214,15 @@ def run(self):
         """
         Run in loop and wait to receive tasks to perform.
         """
-        amount_single_gate = 0
         while True:
             item = self.queue.get()
             if item is None:
                 return
             elif item[0] == SINGLE_GATE:
                 self.apply_single_gate(item[1], item[2])
-                amount_single_gate += 1
             elif item[0] == CONTROLLED_GATE:
                 self.apply_controlled_gate(item[1], item[2], item[3])
             elif item[0] == MEASURE:
-                sys.stdout.flush()
                 self.measure(item[1], item[2])
                 # no qubit left, terminate
                 if len(self.qubits) == 0:
diff --git a/eqsn/shared_dict.py b/eqsn/shared_dict.py
@@ -79,6 +79,11 @@ def get_instance():
             return SharedDict()
         return SharedDict.__instance
 
+    @staticmethod
+    def get_new_instance():
+        SharedDict.__instance = None
+        return SharedDict()
+
     def __init__(self):
         if SharedDict.__instance is not None:
             raise Exception("Call get instance to get this class!")
@@ -90,6 +95,17 @@ def __init__(self):
         self.thread_list = []
         self.queue_list = []
 
+    def get_queues_and_threads_for_ids(self, q_id_list):
+        ret = []
+        self.lock.acquire_read()
+        for q_id in q_id_list:
+            res = self.id_to_queue[q_id]
+            res2 = self.id_to_thread[q_id]
+            if res not in ret:
+                ret.append((res, res2))
+        self.lock.release_read()
+        return ret
+
     def get_queues_for_ids(self, q_id_list):
         ret = []
         self.lock.acquire_read()
@@ -150,6 +166,15 @@ def change_thread_and_queue_of_ids_and_join(self, q_ids, q_id_new_thread):
             self.id_to_queue[q_id] = new_queue
         self.lock.release_write()
 
+    def change_thread_and_queue_of_ids(self, q_ids, q_id_new_thread):
+        self.lock.acquire_write()
+        new_thread = self.id_to_thread[q_id_new_thread]
+        new_queue = self.id_to_queue[q_id_new_thread]
+        for q_id in q_ids:
+            self.id_to_thread[q_id] = new_thread
+            self.id_to_queue[q_id] = new_queue
+        self.lock.release_write()
+
     def send_all_threads(self, msg):
         self.lock.acquire_write()
         for p in self.queue_list:
@@ -161,3 +186,6 @@ def stop_all_threads(self):
         for p in self.thread_list:
             p.join()
         self.lock.release_write()
+
+    def stop_shared_dict(self):
+        SharedDict.__instance = None
diff --git a/eqsn/worker_process.py b/eqsn/worker_process.py
@@ -0,0 +1,150 @@
+import logging
+import threading
+from queue import Queue
+from eqsn.shared_dict import SharedDict
+from eqsn.qubit_thread import SINGLE_GATE, MERGE_SEND, MERGE_ACCEPT, MEASURE,\
+                MEASURE_NON_DESTRUCTIVE, GIVE_QUBITS_AND_TERMINATE, \
+                CONTROLLED_GATE, NEW_QUBIT, ADD_MERGED_QUBITS_TO_DICT, QubitThread
+
+
+class WorkerProcess(object):
+
+    def __init__(self, queue):
+        self.queue = queue
+        # Get a new instance, since their might be one from the old process
+        self.shared_dict = SharedDict.get_new_instance()
+
+    def run(self):
+        """
+        Run in loop and wait to receive tasks to perform.
+        """
+        amount_single_gate = 0
+        while True:
+            item = self.queue.get()
+            if item is None:
+                self.stop_all()
+                return
+            elif item[0] == NEW_QUBIT:
+                self.new_qubit(item[1])
+            elif item[0] == SINGLE_GATE:
+                self.apply_single_gate(item[1], item[2])
+                amount_single_gate += 1
+            elif item[0] == CONTROLLED_GATE:
+                self.apply_controlled_gate(item[1], item[2], item[3])
+            elif item[0] == MEASURE:
+                self.measure(item[1], item[2])
+            elif item[0] == MERGE_ACCEPT:
+                self.merge_accept(item[1], item[2])
+            elif item[0] == MERGE_SEND:
+                self.merge_send(item[1], item[2])
+            elif item[0] == GIVE_QUBITS_AND_TERMINATE:
+                self.give_qubits_and_terminate(item[1], item[2])
+            elif item[0] == MEASURE_NON_DESTRUCTIVE:
+                self.measure_non_destructive(item[1], item[2])
+            elif item[0] == ADD_MERGED_QUBITS_TO_DICT:
+                self.add_merged_qubits_to_thread(item[1], item[2])
+            else:
+                raise ValueError("Command does not exist!")
+
+    def new_qubit(self, q_id):
+        """
+        Creates a new qubit with an id.
+
+        Args:
+            id (String): Id of the new qubit.
+        """
+        q = Queue()
+        thread = QubitThread(q_id, q)
+        p = threading.Thread(target=thread.run, args=())
+        self.shared_dict.set_thread_with_id(q_id, p, q)
+        p.start()
+        logging.debug("Created new qubit with id %s.", q_id)
+
+    def measure(self, q_id, channel):
+        temp_queue = Queue()
+        q = self.shared_dict.get_queues_for_ids([q_id])[0]
+        q.put([MEASURE, q_id, temp_queue])
+        res = temp_queue.get()
+        channel.put(res)
+        self.shared_dict.delete_id_and_check_to_join_thread(q_id)
+
+    def measure_non_destructive(self, q_id, channel):
+        q = self.shared_dict.get_queues_for_ids([q_id])[0]
+        q.put([MEASURE_NON_DESTRUCTIVE, q_id, channel])
+
+    def give_qubits_and_terminate(self, q_id, queue):
+        temp_queue = Queue()
+        q = self.shared_dict.get_queues_for_ids([q_id])[0]
+        q.put([GIVE_QUBITS_AND_TERMINATE, temp_queue])
+        qubits = temp_queue.get()
+        # remove all qubits
+        for c in qubits:
+            self.shared_dict.delete_id_and_check_to_join_thread(c)
+        # send the qubits to the main process
+        queue.put(qubits)
+
+    def add_merged_qubits_to_thread(self, q_id, qubits):
+        q, t = self.shared_dict.get_queues_and_threads_for_ids([q_id])[0]
+        for qubit in qubits:
+            self.shared_dict.set_thread_with_id(qubit, t, q)
+
+    def stop_all(self):
+        """
+        Stops the simulator from running.
+        """
+        self.shared_dict.send_all_threads(None)
+        self.shared_dict.stop_all_threads()
+        self.shared_dict.stop_shared_dict()
+
+    def apply_single_gate(self, gate, q_id):
+        """
+        Applies a single gate to a Qubit.
+        """
+        q = self.shared_dict.get_queues_for_ids([q_id])[0]
+        q.put([SINGLE_GATE, gate, q_id])
+
+    def apply_controlled_gate(self, gate, q_id1, q_id2):
+        """
+        Applies a controlled gate, where the gate is applied to
+        q_id1 and controlled by q_id2.
+
+        Args:
+            q_id1 (String): Id of the Qubit on which the X gate is applied.
+            q_id2 (String): Id of the Qubit which controls the gate.
+        """
+        self.merge_qubits(q_id1, q_id2)
+        q = self.shared_dict.get_queues_for_ids([q_id1])[0]
+        q.put([CONTROLLED_GATE, gate, q_id1, q_id2])
+
+    def merge_send(self, q_id, queue):
+        q = self.shared_dict.get_queues_for_ids([q_id])[0]
+        q.put([MERGE_SEND, queue])
+
+    def merge_accept(self, q_id, queue):
+        q = self.shared_dict.get_queues_for_ids([q_id])[0]
+        q.put([MERGE_ACCEPT, queue])
+
+    def merge_qubits(self, q_id1, q_id2):
+        """
+        Merges two qubits to one process, if they are not already
+        running in the same process.
+
+        Args:
+            q_id1 (String): Id of the Qubit merged into q_id2.
+            q_id2 (String): Id of the Qubit merged with q_id1.
+        """
+        l = self.shared_dict.get_queues_for_ids([q_id1, q_id2])
+        if len(l) == 1:
+            return  # Already merged
+        else:
+            logging.debug("Merge Qubits %s and %s.", q_id1, q_id2)
+            q1 = l[0]
+            q2 = l[1]
+            merge_q = Queue()
+            q1.put([MERGE_SEND, merge_q])
+            q2.put([MERGE_ACCEPT, merge_q])
+            qubits_q = Queue()
+            q1.put([GIVE_QUBITS_AND_TERMINATE, qubits_q])
+            qubits = qubits_q.get()
+            self.shared_dict.change_thread_and_queue_of_ids_and_join(
+                qubits, q_id2)
diff --git a/tests/test_measuring_from_threads.py b/tests/test_measuring_from_threads.py
@@ -11,13 +11,13 @@ def test_measure_from_threads():
     def measure_or_hadamard(id):
         n = random.randrange(10, 100, 1)
         for _ in range(n):
-            time.sleep(0.1)
+            time.sleep(0.05)
             q_sim.H_gate(id)
         print("Finished Hadamard, measure qubit %s!" % id)
         print(q_sim.measure(id))
         print("Finished with Measure!")
 
-    nr_threads = 10
+    nr_threads = 5
     ids = [str(x) for x in range(nr_threads)]
     for id in ids:
         q_sim.new_qubit(id)
@@ -27,7 +27,8 @@ def measure_or_hadamard(id):
             q_sim.cnot_gate(id1, c)
     thread_list = []
     for id in ids:
-        t = threading.Thread(target=measure_or_hadamard, args=(id))
+        print(id)
+        t = threading.Thread(target=measure_or_hadamard, args=(id,))
         t.start()
         thread_list.append(t)
     for t in thread_list:
diff --git a/tests/test_multi_gate.py b/tests/test_multi_gate.py
@@ -1,4 +1,5 @@
 from eqsn import EQSN
+import time
 
 
 def test_epr_creation():
@@ -36,4 +37,5 @@ def test_5_qubits_gate():
 
 if __name__ == "__main__":
     test_epr_creation()
+    time.sleep(0.1)
     test_5_qubits_gate()
diff --git a/tests/test_single_gate.py b/tests/test_single_gate.py
