diff --git a/proton-c/bindings/cpp/src/container_impl.cpp b/proton-c/bindings/cpp/src/container_impl.cpp
index 9cec831..d80f903 100644
--- a/proton-c/bindings/cpp/src/container_impl.cpp
+++ b/proton-c/bindings/cpp/src/container_impl.cpp
@@ -47,8 +47,12 @@
 #include <proton/connection.h>
 #include <proton/handlers.h>
 #include <proton/reactor.h>
+#include <reactor/io.h>
 #include <proton/session.h>
 
+#include <pthread.h>
+#include <unistd.h>
+
 namespace proton {
 
 class container::impl::handler_context {
@@ -143,6 +147,7 @@ container::impl::impl(container& c, const std::string& id, messaging_handler *mh
     // Note: we have just set up the following handlers that see
     // events in this order: messaging_adapter, connector override,
     // the reactor's default globalhandler (pn_iohandler)
+    mt_init();
 }
 
 namespace {
@@ -159,6 +164,7 @@ void close_acceptor(acceptor a) {
 container::impl::~impl() {
     for (acceptors::iterator i = acceptors_.begin(); i != acceptors_.end(); ++i)
         close_acceptor(i->second);
+    mt_cleanup();
 }
 
 // FIXME aconway 2016-06-07: this is not thread safe. It is sufficient for using
@@ -274,42 +280,6 @@ void container::impl::schedule(container& c, int delay, proton_handler *h) {
     schedule(*c.impl_.get(), delay, h);
 }
 
-namespace {
-// Abstract base for timer_handler_std and timer_handler_03
-struct timer_handler : public proton_handler, public void_function0 {
-    void on_timer_task(proton_event& ) PN_CPP_OVERRIDE {
-        (*this)();
-        delete this;
-    }
-    void on_reactor_final(proton_event&) PN_CPP_OVERRIDE {
-        delete this;
-    }
-};
-
-struct timer_handler_03 : public timer_handler {
-    void_function0& func;
-    timer_handler_03(void_function0& f): func(f) {}
-    void operator()() PN_CPP_OVERRIDE { func(); }
-};
-}
-
-void container::impl::schedule(duration delay, void_function0& f) {
-    schedule(*this, delay.milliseconds(), new timer_handler_03(f));
-}
-
-#if PN_CPP_HAS_STD_FUNCTION
-namespace {
-struct timer_handler_std : public timer_handler {
-    std::function<void()> func;
-    timer_handler_std(std::function<void()> f): func(f) {}
-    void operator()() PN_CPP_OVERRIDE { func(); }
-};
-}
-
-void container::impl::schedule(duration delay, std::function<void()> f) {
-    schedule(*this, delay.milliseconds(), new timer_handler_std(f));
-}
-#endif
 
 void container::impl::client_connection_options(const connection_options &opts) {
     client_connection_options_ = opts;
@@ -348,6 +318,7 @@ void container::impl::configure_server_connection(connection &c) {
 }
 
 void container::impl::run() {
+    running_tid = pthread_self();
     do {
         reactor_.run();
     } while (!auto_stop_);
@@ -362,4 +333,256 @@ void container::impl::auto_stop(bool set) {
     auto_stop_ = set;
 }
 
+/*
+ * MT safe schedule.
+ *
+ * Loosely based on the Python binding EventInjector.  We have to make
+ * sure our extra pipe is invisible to the reactor's selectables count
+ * (used to determine reactor final state, see pni_reactor_more).
+ */
+
+/* === simple mutex as used in epoll.c. Only used twice. === */
+typedef pthread_mutex_t pmutex;
+static void pmutex_init(pthread_mutex_t *pm){
+  pthread_mutexattr_t attr;
+
+  pthread_mutexattr_init(&attr);
+  pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ADAPTIVE_NP);
+  if (pthread_mutex_init(pm, &attr)) {
+      throw error("pthread mutex init failure");
+  }
+}
+
+static void pmutex_finalize(pthread_mutex_t *m) { pthread_mutex_destroy(m); }
+static inline void lock(pmutex *m) { pthread_mutex_lock(m); }
+static inline void unlock(pmutex *m) { pthread_mutex_unlock(m); }
+
+/* === similar work and jobs concept as in 0.18 C++ === */
+
+namespace {
+
+union pni_func_t {
+    void_function0 *func_03;  // can't be a reference in a union
+#if PN_CPP_HAS_STD_FUNCTION
+    std::function<void()> std_func;
+#endif
+};
+
+struct work {
+    duration delay;
+    bool is_03;
+    pni_func_t func; 
+
+    work (duration d, void_function0& f): delay(d), is_03(true) { func.func_03 = &f; }
+#if PN_CPP_HAS_STD_FUNCTION
+    work (duration d, std::function<void()> f): delay(d), is_03(false) { func.std_func = f; }
+#endif
+    void operator()() {
+        if (is_03) {
+            (*func.func_03)();
+        }
+#if PN_CPP_HAS_STD_FUNCTION
+        else
+            func.std_func();
+#endif
+    }
+};
+
+}
+
+class mt_scheduler {
+public:
+    mt_scheduler(container &c, pn_reactor_t *r);
+    ~mt_scheduler();
+    void schedule(duration, void_function0&);
+#if PN_CPP_HAS_STD_FUNCTION
+    void schedule(duration, std::function<void()>);
+#endif
+    void tick();
+private:
+    container &container_;
+    pmutex mutex_;
+    pn_socket_t mt_pipe_[2];
+//    pn_io_t *reactor_io_;
+    std::vector<work> jobs_;
+    std::vector<work> delayed_jobs_;
+};
+
+PN_HANDLE(PN_MTSCHED)
+
+namespace {
+/* reactor->selectables count is monitored for detecting FINAL state.
+   This is called before there is any opportunity to make connections
+   or listeners, and just after the mt_selectable is created.  So the
+   count should be exactly one and we decrement it back to zero.  This
+   hack is ugly but hopefully not too dangerous assuming little
+   ongoing development in the pre-proactor universe. */
+bool adjust_reactor_count(pn_reactor_t *r) {
+    struct reactor_copy {
+        pn_record_t *attachments;
+        pn_io_t *io;
+        pn_collector_t *collector;
+        pn_handler_t *global;
+        pn_handler_t *handler;
+        pn_list_t *children;
+        pn_timer_t *timer;
+        pn_socket_t wakeup[2];
+        pn_selectable_t *selectable;
+        pn_event_type_t previous;
+        pn_timestamp_t now;
+        int selectables;
+        int timeout;
+        bool yield;
+        bool stop;
+    } *copy = (struct reactor_copy*) r;
+    if (copy->selectables != 1) return false;
+    copy->selectables = 0;
+    return true;
+}
+
+void mt_sel_readable(pn_selectable_t *mt_sel) {
+    pn_record_t *record = pn_selectable_attachments(mt_sel);
+    mt_scheduler *mts = (mt_scheduler *) pn_record_get(record, PN_MTSCHED);
+    mts->tick();
+}
+
+}
+
+mt_scheduler::mt_scheduler(container &c, pn_reactor_t *r) : container_(c) {
+    pn_selectable_t *mt_sel = pn_reactor_selectable(r);
+    if (!mt_sel || !adjust_reactor_count(r))
+        throw error("MT reactor initialization error");
+    //if (!pn_pipe(reactor_io_, mt_pipe_))
+    if (pipe(mt_pipe_))
+        throw error("MT schedule initialization error");
+
+    pn_record_t *record = pn_selectable_attachments(mt_sel);
+    pn_record_def(record, PN_MTSCHED, PN_VOID);
+    pn_record_set(record, PN_MTSCHED, this);
+    pn_selectable_set_fd(mt_sel, mt_pipe_[0]);
+    pn_selectable_on_readable(mt_sel, mt_sel_readable);
+    pn_selectable_set_reading(mt_sel, true);
+    pn_reactor_update(r, mt_sel);
+    pmutex_init(&mutex_);
+}
+        
+mt_scheduler::~mt_scheduler() {
+    //pn_close(reactor_io_, mt_pipe_[0]);
+    //pn_close(reactor_io_, mt_pipe_[1]);
+    close(mt_pipe_[0]);
+    close(mt_pipe_[1]);
+    pmutex_finalize(&mutex_);
+}
+
+void container::impl::mt_init() {
+    mt_scheduler_ = new mt_scheduler(container_, unwrap(reactor_));
+}
+void container::impl::mt_cleanup() {
+    delete mt_scheduler_;
+}
+
+void mt_scheduler::schedule(duration delay, void_function0& f) {
+    {
+        lock(&mutex_);
+        work w(delay, f);
+        if (delay.milliseconds())
+            delayed_jobs_.push_back(w);
+        else
+            jobs_.push_back(w);
+        unlock(&mutex_);
+    }
+    //pn_write(reactor_io_, mt_pipe_[1], "!", 1);
+    write(mt_pipe_[1], "!", 1);
+}
+
+#if PN_CPP_HAS_STD_FUNCTION
+void mt_scheduler::schedule(duration delay, std::function<foid()> f) {
+    {
+        lock(&mutex_);
+        work w(delay, f);
+        if (delay.milliseconds())
+            delayed_jobs_.push_back(w);
+        else
+            jobs_.push_back(w);
+        unlock(&mutex_);
+    }
+    //pn_write(reactor_io_, mt_pipe_[1], "!", 1);
+    write(mt_pipe_[1], "!", 1);
+}
+
+#endif
+
+void mt_scheduler::tick() {
+    std::vector<work> immediate_jobs;
+    char ignore_buf[512];
+    //pn_read(reactor_io_, mt_pipe_[0], ignore_buf, sizeof(ignore_buf));
+    read(mt_pipe_[0], ignore_buf, sizeof(ignore_buf));
+    {
+        lock(&mutex_);
+        for (std::vector<work>::iterator w = delayed_jobs_.begin(); w != delayed_jobs_.end(); ++w) {
+            // do a normal schedule on the reactor's timer
+            if (w->is_03)
+                container_.schedule(w->delay, *(w->func.func_03));
+#if PN_CPP_HAS_STD_FUNCTION
+            else
+                container_.schedule(w->delay, w->func.std_func);
+#endif
+        }
+        delayed_jobs_.clear();
+        std::swap(immediate_jobs, jobs_);
+        unlock(&mutex_);
+    }
+    
+    for (std::vector<work>::iterator w = immediate_jobs.begin(); w != immediate_jobs.end(); ++w) try {
+        (*w)();
+    } catch (...) {};
+    return;
+}
+
+
+namespace {
+// Abstract base for timer_handler_std and timer_handler_03
+struct timer_handler : public proton_handler, public void_function0 {
+    void on_timer_task(proton_event& ) PN_CPP_OVERRIDE {
+        (*this)();
+        delete this;
+    }
+    void on_reactor_final(proton_event&) PN_CPP_OVERRIDE {
+        delete this;
+    }
+};
+
+
+struct timer_handler_03 : public timer_handler {
+    void_function0& func;
+    timer_handler_03(void_function0& f): func(f) {}
+    void operator()() PN_CPP_OVERRIDE { func(); }
+};
+}
+
+void container::impl::schedule(duration delay, void_function0& f) {
+    if (pthread_equal(pthread_self(), running_tid))
+        schedule(*this, delay.milliseconds(), new timer_handler_03(f));
+    else
+        mt_scheduler_->schedule(delay, f);
+}
+
+#if PN_CPP_HAS_STD_FUNCTION
+namespace {
+struct timer_handler_std : public timer_handler {
+    std::function<void()> func;
+    timer_handler_std(std::function<void()> f): func(f) {}
+    void operator()() PN_CPP_OVERRIDE { func(); }
+};
+}
+
+void container::impl::schedule(duration delay, std::function<void()> f) {
+    if (pthread_equal(pthread_self(), running_tid))
+        schedule(*this, delay.milliseconds(), new timer_handler_std(f));
+    else
+        mt_scheduler_->schedule(delay, f);
+}
+#endif
+
+
 }
diff --git a/proton-c/bindings/cpp/src/include/container_impl.hpp b/proton-c/bindings/cpp/src/include/container_impl.hpp
index 7443150..db0b0d1 100644
--- a/proton-c/bindings/cpp/src/include/container_impl.hpp
+++ b/proton-c/bindings/cpp/src/include/container_impl.hpp
@@ -48,6 +48,7 @@ class connector;
 class acceptor;
 class url;
 class listen_handler;
+class mt_scheduler;
 
 class container::impl {
   public:
@@ -84,6 +85,8 @@ class container::impl {
     static void schedule(impl& ci, int delay, proton_handler *h);
     static void schedule(container& c, int delay, proton_handler *h);
     template <class T> static void set_handler(T s, messaging_handler* h);
+    void mt_init();
+    void mt_cleanup();
 
   private:
     class handler_context;
@@ -103,6 +106,9 @@ class container::impl {
     typedef std::map<std::string, acceptor> acceptors;
     acceptors acceptors_;
     bool auto_stop_;
+    mt_scheduler *mt_scheduler_;
+    pthread_t running_tid;
+    friend class mt_scheduler;
 };
 
 template <class T>
