Merge pull request #7 from tqsd/making_merge_atomar

Making merge atomar

Author: benjione

eqsn/gates.py

@@ -2,7 +2,7 @@
 import logging
 import numpy as np
 from eqsn.qubit_thread import SINGLE_GATE, MERGE_SEND, MERGE_ACCEPT, MEASURE,\
-                MEASURE_NON_DESTRUCTIVE, GIVE_QUBITS_AND_TERMINATE, \
+                MEASURE_NON_DESTRUCTIVE, \
                 CONTROLLED_GATE, NEW_QUBIT, ADD_MERGED_QUBITS_TO_DICT
 from eqsn.shared_dict import SharedDict
 from eqsn.worker_process import WorkerProcess
@@ -19,17 +19,16 @@ 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 = []
+        cpu_count = multiprocessing.cpu_count()
+        self.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_queue_list.append((p, q))
         self.process_picker = ProcessPicker.get_instance(
-            cpu_count, process_queue_list)
+            cpu_count, self.process_queue_list)
 
     def new_qubit(self, q_id):
         """
@@ -47,8 +46,9 @@ def stop_all(self):
         """
         Stops the simulator from running.
         """
-        self.shared_dict.send_all_threads(None)
-        self.shared_dict.stop_all_threads()
+        for p, q in self.process_queue_list:
+            q.put(None)
+            p.join()
         self.shared_dict.stop_shared_dict()
         self.process_picker.stop_process_picker()
 
@@ -153,18 +153,20 @@ def merge_qubits(self, q_id1, q_id2):
         if len(l) == 1:
             return  # Already merged
         else:
+            # Block the dictionary, that noone can send commands to the qubits,
             logging.debug("Merge Qubits %s and %s.", q_id1, q_id2)
+            self.shared_dict.block_shared_dict()
             q1 = l[0]
             q2 = l[1]
             merge_q = self.manager.Queue()
-            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, q_id1, qubits_q])
+            q1.put([MERGE_SEND, q_id1, merge_q, qubits_q])
+            q2.put([MERGE_ACCEPT, q_id2, merge_q])
             qubits = qubits_q.get()
             q2.put([ADD_MERGED_QUBITS_TO_DICT, q_id2, qubits])
-            self.shared_dict.change_thread_and_queue_of_ids(
+            self.shared_dict.change_thread_and_queue_of_ids_nonblocking(
                 qubits, q_id2)
+            self.shared_dict.release_shared_dict()
 
     def cnot_gate(self, q_id1, q_id2):
         """

eqsn/qubit_thread.py

@@ -10,7 +10,6 @@
 MEASURE = 3
 MERGE_ACCEPT = 4
 MERGE_SEND = 5
-GIVE_QUBITS_AND_TERMINATE = 6
 MEASURE_NON_DESTRUCTIVE = 7
 NEW_QUBIT = 8
 ADD_MERGED_QUBITS_TO_DICT = 9
@@ -115,19 +114,13 @@ def merge_accept(self, channel):
         self.qubit = np.kron(self.qubit, vector)
         logging.debug("Qubit Thread merged, new qubits are %r", self.qubits)
 
-    def merge_send(self, channel):
+    def merge_send(self, channel, chanel2):
         """
         Send own process data to another process and suicide.
         """
         channel.put(dp(self.qubits))
         channel.put(dp(self.qubit))
-        return
-
-    def send_qubits(self, channel):
-        """
-        Send which qubits are in this process over a channel.
-        """
-        channel.put(dp(self.qubits))
+        chanel2.put(dp(self.qubits))
         return
 
     def measure_non_destructive(self, q_id, ret_channel):
@@ -145,7 +138,11 @@ def measure_non_destructive(self, q_id, ret_channel):
         if after > 0:
             measure_vec = np.kron(measure_vec, np.ones(2 ** after))
         pr_0 = np.multiply(measure_vec, self.qubit)
-        pr_0 = abs(np.dot(pr_0, pr_0).real)
+        pr_0 = abs(np.dot(pr_0, pr_0))
+        if pr_0 > 1.0:
+            pr_0 = 1.0
+        elif pr_0 < 0.0:
+            pr_0 = 0.0
         meas_res = np.random.binomial(1, 1.0 - pr_0)
         reduction_mat = None
         if meas_res == 0:
@@ -183,7 +180,11 @@ def measure(self, q_id, ret_channel):
         if after > 0:
             measure_vec = np.kron(measure_vec, np.ones(2 ** after))
         pr_0 = np.multiply(measure_vec, self.qubit)
-        pr_0 = abs(np.dot(pr_0, pr_0).real)
+        pr_0 = abs(np.dot(pr_0, pr_0))
+        if pr_0 > 1.0:
+            pr_0 = 1.0
+        elif pr_0 < 0.0:
+            pr_0 = 0.0
         meas_res = np.random.binomial(1, 1.0 - pr_0)
         reduction_mat = None
         if meas_res == 0:
@@ -230,9 +231,7 @@ def run(self):
             elif item[0] == MERGE_ACCEPT:
                 self.merge_accept(item[1])
             elif item[0] == MERGE_SEND:
-                self.merge_send(item[1])
-            elif item[0] == GIVE_QUBITS_AND_TERMINATE:
-                self.send_qubits(item[1])
+                self.merge_send(item[1], item[2])
                 return
             elif item[0] == MEASURE_NON_DESTRUCTIVE:
                 self.measure_non_destructive(item[1], item[2])

eqsn/shared_dict.py

@@ -95,6 +95,12 @@ def __init__(self):
         self.thread_list = []
         self.queue_list = []
 
+    def block_shared_dict(self):
+        self.lock.acquire_write()
+
+    def release_shared_dict(self):
+        self.lock.release_write()
+
     def get_queues_and_threads_for_ids(self, q_id_list):
         ret = []
         self.lock.acquire_read()
@@ -175,6 +181,13 @@ def change_thread_and_queue_of_ids(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_nonblocking(self, q_ids, q_id_new_thread):
+        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
+
     def send_all_threads(self, msg):
         self.lock.acquire_write()
         for p in self.queue_list:

eqsn/worker_process.py

@@ -1,9 +1,10 @@
 import logging
 import threading
+import os
 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, \
+                MEASURE_NON_DESTRUCTIVE, \
                 CONTROLLED_GATE, NEW_QUBIT, ADD_MERGED_QUBITS_TO_DICT, QubitThread
 
 
@@ -36,9 +37,7 @@ def run(self):
             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])
+                self.merge_send(item[1], item[2], item[3])
             elif item[0] == MEASURE_NON_DESTRUCTIVE:
                 self.measure_non_destructive(item[1], item[2])
             elif item[0] == ADD_MERGED_QUBITS_TO_DICT:
@@ -72,17 +71,6 @@ 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:
@@ -116,9 +104,16 @@ def apply_controlled_gate(self, gate, 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):
+    def merge_send(self, q_id, queue, queue2):
+        temp_queue = Queue()
         q = self.shared_dict.get_queues_for_ids([q_id])[0]
-        q.put([MERGE_SEND, queue])
+        q.put([MERGE_SEND, queue, 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
+        queue2.put(qubits)
 
     def merge_accept(self, q_id, queue):
         q = self.shared_dict.get_queues_for_ids([q_id])[0]
@@ -141,10 +136,9 @@ def merge_qubits(self, 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])
+            q1.put([MERGE_SEND, merge_q, qubits_q])
+            q2.put([MERGE_ACCEPT, merge_q])
             qubits = qubits_q.get()
             self.shared_dict.change_thread_and_queue_of_ids_and_join(
                 qubits, q_id2)

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.05)
+            # 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 = 5
+    nr_threads = 10
     ids = [str(x) for x in range(nr_threads)]
     for id in ids:
         q_sim.new_qubit(id)

tests/test_merging_qubits.py

@@ -9,6 +9,7 @@ def test_merge():
     q_sim.new_qubit(id2)
     q_sim.merge_qubits(id1, id2)
     q_sim.stop_all()
+    print("merging was succesfull")
 
 
 if __name__ == "__main__":

tests/test_multi_gate.py

@@ -37,5 +37,5 @@ def test_5_qubits_gate():
 
 if __name__ == "__main__":
     test_epr_creation()
-    time.sleep(0.1)
+    time.sleep(0.05)
     test_5_qubits_gate()

tests/test_non_destructive_measurement.py

@@ -12,6 +12,8 @@ def test_non_destructive_measurement():
     assert m == m2
     print("Test was successfull!")
     q_sim.stop_all()
+    print("Stopped succesfully!")
+    exit(0)
 
 
 if __name__ == "__main__":

tests/test_single_gate.py

@@ -118,4 +118,4 @@ def test_measure():
                 test_measure]
     for func in test_list:
         func()
-        time.sleep(0.01)
+        time.sleep(0.05)