基本完成异步框架, 还有内存泄露.

Author: 朱胜

src/couv/channel.h

@@ -28,6 +28,10 @@ class channel
     bool is_closed() {
         return m_closed;
     }
+    
+    operator bool() const {
+        return !m_closed;
+    }
 
     void close() {
         m_closed = true;
@@ -42,13 +46,15 @@ class channel
         m_sem.signal();
     }
 
-    void receive(T& t) {
-        if(m_closed)
-            return;
+    channel<T>& receive(T& t) {
+        if(m_closed) {
+            return *this;
+        }
 
         m_sem.wait();
         t = m_queue.front();
         m_queue.pop_front();
+        return *this;
     }
 
     T receive() {

src/couv/coroutine.h

@@ -22,6 +22,9 @@ class coroutine_base;
 // customer
 class coroutine: public coroutine_base
 {
+public:
+    typedef void sign_type();
+    typedef std::function<sign_type> func_type;
 public:
     coroutine(func_type f);
 

src/couv/coroutine_base.cpp

@@ -6,7 +6,6 @@
  */
 #include "couv/coroutine_base.h"
 #include "couv/scheduler.h"
-#include "couv/scheduler.h"
 #include "log/Logger.h"
 
 namespace couv
@@ -40,8 +39,13 @@ void coroutine_base::resume_coroutine(coroutine_ptr other)
     m_active = false;
     other->m_active = true;
 
-    current_scheduler->set_current(other);
+    scheduler_t::self()->set_current(other);
     swapcontext(&m_ctx, &other->m_ctx);
 }
 
+coroutine_ptr coroutine_base::self()
+{
+    return scheduler_t::self()->current();
+}
+
 } /* namespace coroutine */

src/couv/coroutine_base.h

@@ -19,10 +19,6 @@ namespace couv
 
 class coroutine_base
 {
-public:
-    typedef void sign_type();
-    typedef std::function<sign_type> func_type;
-
 public:
     coroutine_base();
     virtual ~coroutine_base();
@@ -35,8 +31,10 @@ class coroutine_base
     bool is_active()const  { return m_active;  }
     bool is_blocked()const { return m_blocked; }
     bool is_done()const    { return m_done;    }
-    void set_blocked(bool block)  { m_blocked = block; }
+    void set_blocked(bool block)    { m_blocked = block; }
     void set_delegate(delegate_t *d){ m_delegate = d;    }
+
+    static coroutine_ptr self();
 protected:
     bool m_active;
     bool m_blocked;

src/couv/lock.cpp

@@ -21,7 +21,7 @@ lock_t::~lock_t()
 
 void lock_t::lock()
 {
-    coroutine_ptr current = current_coroutine;
+    coroutine_ptr current = coroutine_base::self();
     if(m_flag) { // wait
         if(m_owner != current && m_queue.find(current) == m_queue.end()) {
             current->set_blocked(true);
@@ -50,7 +50,7 @@ bool lock_t::try_lock()
 {
     if(!m_flag) {
         m_flag = true;
-        m_owner = current_coroutine;
+        m_owner = coroutine_base::self();
         return true;
     }
     return false;

src/couv/scheduler.cpp

@@ -11,16 +11,17 @@
 
 namespace couv
 {
-//static  scheduler_t local_scheduler;
-//scheduler_t* current_scheduler = &local_scheduler;
-scheduler_t* current_scheduler = nullptr;
+
+scheduler_t* scheduler_t::s_instance = nullptr;
+
 
 scheduler_t::scheduler_t(bool install)
+        :m_stop(false)
 {
     m_root = std::make_shared<coroutine_base>();
     m_current = m_root;
     if(install) {
-        current_scheduler = this;
+        s_instance = this;
     }
 }
 
@@ -29,24 +30,47 @@ scheduler_t::~scheduler_t()
     logAssert(m_current == m_root);
 }
 
-coroutine_ptr scheduler_t::add(coroutine_base::func_type&& f)
-{
-    return add(coroutine_ptr(new coroutine(f)));
-}
-
-coroutine_ptr scheduler_t::add(coroutine_ptr r)
+bool scheduler_t::add(coroutine_ptr r)
 {
     logAssert(r != nullptr);
 
     if(m_root == r || r->is_done()) {
-        return r;
+        return false;
     }
 
     logDebug("add %d", r->id());
     m_queue.push_back(r);
+    return true;
+}
+
+coroutine_ptr scheduler_t::add(const func_type& f)
+{
+    coroutine_ptr r = std::make_shared<coroutine>(f);
+    add(r);
     return r;
 }
 
+coroutine_ptr scheduler_t::add(func_type&& f)
+{
+    coroutine_ptr r = std::make_shared<coroutine>(std::forward<func_type>(f));
+    add(r);
+    return r;
+}
+    
+scheduler_t& scheduler_t::operator<<(func_type&& f)
+{
+    coroutine_ptr r = std::make_shared<coroutine>(std::forward<func_type>(f));
+    add(r);
+    return *this;
+}
+    
+scheduler_t& scheduler_t::operator<<(const func_type& f)
+{
+    coroutine_ptr r = std::make_shared<coroutine>(f);
+    add(r);
+    return *this;
+}
+    
 void scheduler_t::remove(coroutine_ptr r)
 {
     logDebug("remove %d", r->id());
@@ -85,7 +109,7 @@ void scheduler_t::run()
 {
     //logAssert(current_scheduler == nullptr);
 
-    current_scheduler = this;
+    install();
 
     while(!m_queue.empty())
     {
@@ -103,16 +127,24 @@ void scheduler_t::run()
 
         if(!found)
         {
-            logWarn("all coroutine blocked \n");
-            exit(1);
+            if(!has_more())
+            {
+                break;
+            }
         }
         else
         {
             logDebug("OK!");
         }
     }
 
-    current_scheduler = nullptr;
+    s_instance = nullptr;
+}
+    
+bool scheduler_t::has_more()
+{
+    logWarn("all coroutine blocked \n");
+    return false;
 }
 
 void scheduler_t::on_start(coroutine_base* r)

src/couv/scheduler.h

@@ -10,20 +10,18 @@
 
 #include <deque>
 
-#include "couv/coroutine_base.h"
 #include "couv/delegate.h"
 #include "log/Logger.h"
 
-#define current_coroutine couv::current_scheduler->current()
-
 #define yield(...) do{ \
-    logDebug("yield from %d", current_coroutine->id()); \
-    couv::current_scheduler->yield_coroutine( __VA_ARGS__ ); \
+    logDebug("yield from %d", couv::coroutine_base::self()->id()); \
+    couv::scheduler_t::self()->yield_coroutine( __VA_ARGS__ ); \
     } while(0)
 
-#define resume couv::current_scheduler->resume
+#define resume couv::scheduler_t::self()->resume
 
-#define go (*couv::current_scheduler)<<[&]
+#define go (*couv::scheduler_t::self())<<[&]
+#define goo (*couv::scheduler_t::self())<<
 
 namespace couv
 {
@@ -33,40 +31,54 @@ namespace couv
  */
 class scheduler_t: public delegate_t
 {
+public:
+    typedef void sign_type();
+    typedef std::function<sign_type> func_type;
 public:
     scheduler_t(bool install=true);
     virtual ~scheduler_t();
+    virtual bool add(coroutine_ptr r);
+    virtual void remove(coroutine_ptr r);
     virtual void run();
-
-    coroutine_ptr add(coroutine_base::func_type&& f);
-    coroutine_ptr add(coroutine_ptr r);
-    void remove(coroutine_ptr r);
-
+    
     void yield_coroutine(coroutine_ptr r);
     void yield_coroutine();
+    
+    coroutine_ptr add(const func_type& f);
+    coroutine_ptr add(func_type&& f);
+    scheduler_t& operator<<(const func_type& f);
+    scheduler_t& operator<<(func_type&& f);
 
     void set_current(coroutine_ptr r) { m_current = r; }
     coroutine_ptr current(){ return m_current; }
-    
-    scheduler_t& operator<<(coroutine_base::func_type&& f){
-        add(std::move(f));
-        return *this;
+
+    void stop()         { m_stop = true; }
+    bool is_stop()const { return m_stop; }
+
+    void install()           { s_instance = this; }
+    bool is_installed()const { return s_instance == this; }
+
+    static scheduler_t* self()
+    {
+        logAssert(s_instance != nullptr);
+        return s_instance;
     }
 
 protected:
-    void resume_coroutine(coroutine_ptr r);
+    virtual void resume_coroutine(coroutine_ptr r);
+    virtual bool has_more();
 
     virtual void on_start(coroutine_base* r);
     virtual void on_stop(coroutine_base* r);
 
 protected:
+    static scheduler_t* s_instance;
+    bool m_stop;
     coroutine_ptr m_root;
     coroutine_ptr m_current;
     std::deque<coroutine_ptr> m_queue;
 };
 
-extern scheduler_t* current_scheduler;
-
 } /* namespace coroutine */
 
 #endif /* SRC_COUV_SCHEDULER_H_ */

src/couv/sem.cpp

@@ -23,8 +23,8 @@ sem_t::~sem_t()
 void sem_t::wait()
 {
     if(--m_counter < 0) {
-        current_coroutine->set_blocked(true);
-        m_queue.push_back(current_coroutine);
+        coroutine_base::self()->set_blocked(true);
+        m_queue.push_back(coroutine_base::self());
         yield();
     }
 }