QPID-4967 - Bug fixes, added a lock to serialize calls into the Python interpreter.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1529307 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 28 insertions, 6 deletions

diff --git a/extras/dispatch/include/qpid/dispatch/python_embedded.h b/extras/dispatch/include/qpid/dispatch/python_embedded.h
index 644885bbd7..aad0453c24 100644
--- a/extras/dispatch/include/qpid/dispatch/python_embedded.h
+++ b/extras/dispatch/include/qpid/dispatch/python_embedded.h
@@ -71,4 +71,11 @@ void dx_py_to_composed(PyObject *value, dx_composed_field_t *field);
  */
 PyObject *dx_field_to_py(dx_parsed_field_t *field);
 
+/**
+ * These are temporary and will eventually be replaced by having an internal python
+ * work queue that feeds a dedicated embedded-python thread.
+ */
+void dx_python_lock();
+void dx_python_unlock();
+
 #endif
diff --git a/extras/dispatch/python/qpid/dispatch/router/router_engine.py b/extras/dispatch/python/qpid/dispatch/router/router_engine.py
index 02f0315104..03bf0638a5 100644
--- a/extras/dispatch/python/qpid/dispatch/router/router_engine.py
+++ b/extras/dispatch/python/qpid/dispatch/router/router_engine.py
@@ -176,6 +176,7 @@ class RouterEngine:
         This is the IoAdapter message-receive handler
         """
         try:
+            #self.log(LOG_DEBUG, "Raw Receive: mp=%r body=%r link_id=%r" % (message_properties, body, link_id))
             self.handleControlMessage(message_properties['opcode'], body, link_id)
         except Exception, e:
             self.log(LOG_ERROR, "Exception in raw message processing: properties=%r body=%r exception=%r" %
diff --git a/extras/dispatch/python/qpid/dispatch/router/routing.py b/extras/dispatch/python/qpid/dispatch/router/routing.py
index 1097d8fbba..a4b3e5484a 100644
--- a/extras/dispatch/python/qpid/dispatch/router/routing.py
+++ b/extras/dispatch/python/qpid/dispatch/router/routing.py
@@ -48,13 +48,13 @@ class RoutingTableEngine(object):
             self.container.router_adapter.set_next_hop(mb_id, mb_nh)
 
 
-    def valid_origins_changes(self, valid_origins):
+    def valid_origins_changed(self, valid_origins):
         for _id, vo in valid_origins.items():
             mb_id = self.node_tracker.maskbit_for_node(_id)
             mb_vo = []
             for o in vo:
                 mb_vo.append(self.node_tracker.maskbit_for_node(o))
-            self.container.router_adapted.set_valid_origins(mb_id, mb_vo)
+            self.container.router_adapter.set_valid_origins(mb_id, mb_vo)
 
 
     def get_next_hops(self):
diff --git a/extras/dispatch/src/hash.c b/extras/dispatch/src/hash.c
index 49f79925f6..a0d93f7567 100644
--- a/extras/dispatch/src/hash.c
+++ b/extras/dispatch/src/hash.c
@@ -270,7 +270,6 @@ dx_error_t hash_remove_by_handle(hash_t *h, hash_handle_t *handle)
     DEQ_REMOVE(handle->bucket->items, handle->item);
     free_hash_item_t(handle->item);
     h->size--;
-    free_hash_handle_t(handle);
     return DX_ERROR_NONE;
 }
 
diff --git a/extras/dispatch/src/python_embedded.c b/extras/dispatch/src/python_embedded.c
index c7c8f82dfc..6b235c7406 100644
--- a/extras/dispatch/src/python_embedded.c
+++ b/extras/dispatch/src/python_embedded.c
@@ -452,6 +452,7 @@ static void dx_io_rx_handler(void *context, dx_message_t *msg, int link_id)
         return;
     }
 
+    sys_mutex_lock(lock);
     PyObject *pAP   = dx_field_to_py(ap_map);
     PyObject *pBody = dx_field_to_py(body_map);
 
@@ -465,6 +466,7 @@ static void dx_io_rx_handler(void *context, dx_message_t *msg, int link_id)
     if (pValue) {
         Py_DECREF(pValue);
     }
+    sys_mutex_unlock(lock);
 }
 
 
@@ -658,3 +660,14 @@ static void dx_python_setup()
         dispatch_module = m;
     }
 }
+
+void dx_python_lock()
+{
+    sys_mutex_lock(lock);
+}
+
+void dx_python_unlock()
+{
+    sys_mutex_unlock(lock);
+}
+
diff --git a/extras/dispatch/src/router_pynode.c b/extras/dispatch/src/router_pynode.c
index 3db9d38a8b..d6286614df 100644
--- a/extras/dispatch/src/router_pynode.c
+++ b/extras/dispatch/src/router_pynode.c
@@ -272,12 +272,12 @@ static PyObject* dx_set_valid_origins(PyObject *self, PyObject *args)
         return 0;
     }
 
-    Py_ssize_t        origin_count = PyTuple_Size(origin_list);
+    Py_ssize_t        origin_count = PyList_Size(origin_list);
     dx_router_node_t *rnode        = router->routers_by_mask_bit[router_maskbit];
     int               maskbit;
 
     for (idx = 0; idx < origin_count; idx++) {
-        maskbit = PyInt_AS_LONG(PyTuple_GetItem(origin_list, idx));
+        maskbit = PyInt_AS_LONG(PyList_GetItem(origin_list, idx));
 
         if (maskbit >= dx_bitmask_width() || maskbit < 0) {
             PyErr_SetString(PyExc_Exception, "Origin bit mask out of range");
@@ -292,7 +292,7 @@ static PyObject* dx_set_valid_origins(PyObject *self, PyObject *args)
 
     dx_bitmask_clear_all(rnode->valid_origins);
     for (idx = 0; idx < origin_count; idx++) {
-        maskbit = PyInt_AS_LONG(PyTuple_GetItem(origin_list, idx));
+        maskbit = PyInt_AS_LONG(PyList_GetItem(origin_list, idx));
         dx_bitmask_set_bit(rnode->valid_origins, maskbit);
     }
 
@@ -534,6 +534,7 @@ void dx_pyrouter_tick(dx_router_t *router)
     PyObject *pValue;
 
     if (router->pyTick) {
+        dx_python_lock();
         pArgs  = PyTuple_New(0);
         pValue = PyObject_CallObject(router->pyTick, pArgs);
         if (PyErr_Occurred()) {
@@ -543,6 +544,7 @@ void dx_pyrouter_tick(dx_router_t *router)
         if (pValue) {
             Py_DECREF(pValue);
         }
+        dx_python_unlock();
     }
 }