TS-3235: fix crash problem caused by sync problem in PluginVC.

This closes apache#164.

Author: zouyu

lib/atscppapi/src/InterceptPlugin.cc

@@ -112,7 +112,7 @@ void destroyCont(InterceptPlugin::State *state);
 
 InterceptPlugin::InterceptPlugin(Transaction &transaction, InterceptPlugin::Type type) : TransactionPlugin(transaction)
 {
-  TSCont cont = TSContCreate(handleEvents, TSMutexCreate());
+  TSCont cont = TSContCreate(handleEvents, NULL);
   state_ = new State(cont, this);
   TSContDataSet(cont, state_);
   TSHttpTxn txn = static_cast<TSHttpTxn>(transaction.getAtsHandle());
@@ -137,6 +137,7 @@ InterceptPlugin::~InterceptPlugin()
 bool
 InterceptPlugin::produce(const void *data, int data_size)
 {
+  ScopedSharedMutexLock lock(getMutex());
   if (!state_->net_vc_) {
     LOG_ERROR("Intercept not operational");
     return false;

lib/atscppapi/src/Makefile.am

@@ -48,7 +48,8 @@ libatscppapi_la_SOURCES = GlobalPlugin.cc \
 			  GzipDeflateTransformation.cc \
 			  GzipInflateTransformation.cc \
 			  AsyncTimer.cc \
-                          InterceptPlugin.cc
+			  InterceptPlugin.cc \
+			  Mutex.cc
 
 library_includedir=$(includedir)/atscppapi
 base_include_folder = $(top_srcdir)/$(subdir)/include/atscppapi

lib/atscppapi/src/Mutex.cc

@@ -0,0 +1,34 @@
+/**
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+ */
+
+/**
+ * @file Mutex.cc
+ */
+
+#include "atscppapi/Mutex.h"
+
+#include <ts/ts.h>
+
+atscppapi::ScopedContinuationLock::ScopedContinuationLock(TSCont contp) : mutex_(TSContMutexGet(contp))
+{
+  TSMutexLock(mutex_);
+}
+atscppapi::ScopedContinuationLock::~ScopedContinuationLock()
+{
+  TSMutexUnlock(mutex_);
+}

lib/atscppapi/src/include/atscppapi/Mutex.h

@@ -29,6 +29,10 @@
 #include <atscppapi/noncopyable.h>
 #include <atscppapi/shared_ptr.h>
 
+// Import in name only.
+typedef struct tsapi_mutex *TSMutex;
+typedef struct tsapi_cont *TSCont;
+
 namespace atscppapi
 {
 /**
@@ -252,6 +256,30 @@ class ScopedSharedMutexTryLock : noncopyable
   bool has_lock_;
 };
 
+/**
+ * @brief Lock a TS Continuation by acquiring and releasing its lock in the current scope.
+ *
+ * This is an RAII implementation which will lock a mutex for the continuation at the declaration of an instance
+ * and unlock it when the instance goes out of scope.
+ */
+class ScopedContinuationLock : noncopyable
+{
+public:
+  /**
+   * Create the scoped mutex lock, once this object is constructed the lock will be held by the thread.
+   * @param cont The TS continuation.
+   */
+  explicit ScopedContinuationLock(TSCont contp);
+
+  /**
+   * Unlock the mutex.
+   */
+  ~ScopedContinuationLock();
+
+private:
+  TSMutex mutex_;
+};
+
 } /* atscppapi */
 
 

proxy/PluginVC.cc

@@ -293,7 +293,9 @@ PluginVC::reenable(VIO *vio)
 {
   ink_assert(!closed);
   ink_assert(magic == PLUGIN_VC_MAGIC_ALIVE);
-  ink_assert(vio->mutex->thread_holding == this_ethread());
+
+  Ptr<ProxyMutex> sm_mutex = vio->mutex;
+  SCOPED_MUTEX_LOCK(lock, sm_mutex, this_ethread());
 
   Debug("pvc", "[%u] %s: reenable %s", core_obj->id, PVC_TYPE, (vio->op == VIO::WRITE) ? "Write" : "Read");
 
@@ -313,34 +315,26 @@ PluginVC::reenable_re(VIO *vio)
 {
   ink_assert(!closed);
   ink_assert(magic == PLUGIN_VC_MAGIC_ALIVE);
-  ink_assert(vio->mutex->thread_holding == this_ethread());
 
   Debug("pvc", "[%u] %s: reenable_re %s", core_obj->id, PVC_TYPE, (vio->op == VIO::WRITE) ? "Write" : "Read");
 
-  MUTEX_TRY_LOCK(lock, this->mutex, this_ethread());
-  if (!lock.is_locked()) {
-    if (vio->op == VIO::WRITE) {
-      need_write_process = true;
-    } else {
-      need_read_process = true;
-    }
-    setup_event_cb(PVC_LOCK_RETRY_TIME, &sm_lock_retry_event);
-    return;
-  }
+  SCOPED_MUTEX_LOCK(lock, this->mutex, this_ethread());
 
-  reentrancy_count++;
+  ++reentrancy_count;
 
   if (vio->op == VIO::WRITE) {
     ink_assert(vio == &write_state.vio);
+    need_write_process = true;
     process_write_side(false);
   } else if (vio->op == VIO::READ) {
     ink_assert(vio == &read_state.vio);
+    need_read_process = true;
     process_read_side(false);
   } else {
     ink_release_assert(0);
   }
 
-  reentrancy_count--;
+  --reentrancy_count;
 
   // To process the close, we need the lock
   //   for the PluginVC.  Schedule an event
@@ -353,31 +347,20 @@ PluginVC::reenable_re(VIO *vio)
 void
 PluginVC::do_io_close(int /* flag ATS_UNUSED */)
 {
-  ink_assert(closed == false);
+  ink_assert(!closed);
   ink_assert(magic == PLUGIN_VC_MAGIC_ALIVE);
 
   Debug("pvc", "[%u] %s: do_io_close", core_obj->id, PVC_TYPE);
 
-  if (reentrancy_count > 0) {
-    // Do nothing since dealloacting ourselves
-    //  now will lead to us running on a dead
-    //  PluginVC since we are being called
-    //  reentrantly
+  SCOPED_MUTEX_LOCK(lock, mutex, this_ethread());
+  if (!closed) { // if already closed, need to do nothing.
     closed = true;
-    return;
-  }
 
-  MUTEX_TRY_LOCK(lock, mutex, this_ethread());
-
-  if (!lock.is_locked()) {
-    setup_event_cb(PVC_LOCK_RETRY_TIME, &sm_lock_retry_event);
-    closed = true;
-    return;
-  } else {
-    closed = true;
+    // If re-entered then that earlier handler will clean up, otherwise set up a ping
+    // to drive that process (too dangerous to do it here).
+    if (reentrancy_count <= 0)
+      setup_event_cb(0, &sm_lock_retry_event);
   }
-
-  process_close();
 }
 
 void