X WatchLink -> refcounted class

wcfs/internal/_wcfs.pyx

@@ -23,7 +23,7 @@
 
 # XXX doc
 
-from golang cimport error
+from golang cimport error, refptr
 from golang cimport context
 
 from libcpp cimport nullptr_t, nullptr as nil
@@ -33,18 +33,24 @@ cdef extern from *:
     ctypedef bint cbool "bool"
 
 cdef extern from "wcfs_watchlink.h" nogil:
-    cppclass WatchLink:
+    cppclass _WatchLink:
         error close()
         #error recvReq(ctx, PinReq *rx_into)
         pair[string, error] sendReq(context.Context ctx, const string &req)
 
-ctypedef WatchLink *pWatchLink # https://github.com/cython/cython/issues/534
+    cppclass WatchLink (refptr[_WatchLink]):
+        # WatchLink.X = WatchLink->X in C++
+        error               close       "_ptr()->close"     ()
+        pair[string, error] sendReq     "_ptr()->sendReq"   (context.Context ctx, const string &req)
+
+#ctypedef WatchLink *pWatchLink # https://github.com/cython/cython/issues/534
 
 cdef extern from "wcfs.h" nogil:
     cppclass WCFS:
         string  mountpoint
 
-        pair[pWatchLink, error] _openwatch()    # XXX pair instead of tuple
+        #pair[pWatchLink, error] _openwatch()    # XXX pair instead of tuple
+        pair[WatchLink, error] _openwatch()    # XXX pair instead of tuple
 
 
 cdef class PyWCFS:
@@ -57,7 +63,7 @@ cdef class PyWCFS:
             pywc.wc.mountpoint = v
 
 cdef class PyWatchLink:
-    cdef WatchLink *wlink
+    cdef WatchLink wlink
 
     def __init__(PyWatchLink pywlink, PyWCFS pywc):
         with nogil:
@@ -68,7 +74,9 @@ cdef class PyWatchLink:
         if err != nil:
             raise RuntimeError(err.Error())     # XXX exc class?
 
-    # XXX __dealloc__
+    def __dealloc__(PyWatchLink pywlink):
+        pywlink.wlink = nil
+
 
     def close(PyWatchLink pywlink):
         with nogil:
@@ -103,11 +111,14 @@ from golang cimport topyexc
 
 cdef nogil:
 
-    pair[pWatchLink, error] wcfs_openwatch_pyexc(WCFS *wcfs)    except +topyexc:
+    #pair[pWatchLink, error] wcfs_openwatch_pyexc(WCFS *wcfs)    except +topyexc:
+    pair[WatchLink, error] wcfs_openwatch_pyexc(WCFS *wcfs)    except +topyexc:
         return wcfs._openwatch()
 
-    error wlink_close_pyexc(WatchLink *wlink)                   except +topyexc:
+    #error wlink_close_pyexc(WatchLink *wlink)                   except +topyexc:
+    error wlink_close_pyexc(WatchLink wlink)                   except +topyexc:
         return wlink.close()
 
-    pair[string, error] wlink_sendReq_pyexc(WatchLink *wlink, context.Context ctx, const string &req)   except +topyexc:
+    #pair[string, error] wlink_sendReq_pyexc(WatchLink *wlink, context.Context ctx, const string &req)   except +topyexc:
+    pair[string, error] wlink_sendReq_pyexc(WatchLink wlink, context.Context ctx, const string &req)   except +topyexc:
         return wlink.sendReq(ctx, req)

wcfs/internal/wcfs.h

@@ -34,7 +34,7 @@ using std::pair;
 #include "wcfs_misc.h"
 
 struct Conn;
-struct WatchLink;
+struct _WatchLink;
 
 
 // WCFS represents filesystem-level connection to wcfs server.
@@ -42,10 +42,10 @@ struct WatchLink;
 struct WCFS {
     string  mountpoint;
 
-    tuple<Conn*, error>         connect(zodb::Tid at);
-    string                      _path(const string &obj);
-    tuple<os::File, error>      _open(const string &path, int flags=O_RDONLY);
-    pair<WatchLink*, error>     _openwatch();
+    tuple<Conn*, error>                 connect(zodb::Tid at);
+    string                              _path(const string &obj);
+    tuple<os::File, error>              _open(const string &path, int flags=O_RDONLY);
+    pair<refptr<_WatchLink>, error>     _openwatch();
 };
 
 

wcfs/internal/wcfs_virtmem.cpp

@@ -51,7 +51,6 @@ static error mmap_into_ro(void *addr, size_t size, const os::File &f, off_t offs
 struct Conn;
 struct _File;
 struct _Mapping;
-struct WatchLink;
 struct PinReq;
 
 // Conn represents logical connection that provides view of data on wcfs
@@ -61,7 +60,7 @@ struct PinReq;
 struct Conn {
     WCFS        *_wc;
     zodb::Tid   at;
-    WatchLink   *_wlink;
+    WatchLink   _wlink;
 
     sync::Mutex              _filemu;
     dict<zodb::Oid, _File*>  _filetab; // {} foid -> _file
@@ -113,7 +112,7 @@ tuple<Conn*, error> WCFS::connect(zodb::Tid at) {
     WCFS *wc = this;
     // XXX err ctx
 
-    WatchLink *wlink;
+    WatchLink wlink;
     error err;
     tie(wlink, err) = wc->_openwatch();
     if (err != nil)

wcfs/internal/wcfs_watchlink.cpp

@@ -23,9 +23,15 @@
 
 #include <string.h>
 
+_WatchLink::_WatchLink()    {}
+_WatchLink::~_WatchLink()   {}
+void _WatchLink::decref() {
+    if (__decref())
+        delete this;
+}
 
 // _openwatch opens new watch link on wcfs.
-pair<WatchLink*, error> WCFS::_openwatch() {
+pair<WatchLink, error> WCFS::_openwatch() {
     WCFS *wc = this;
 
     // head/watch handle.
@@ -33,9 +39,9 @@ pair<WatchLink*, error> WCFS::_openwatch() {
     error err;
     tie(f, err) = wc->_open("head/watch", O_RDWR);
     if (err != nil)
-        return make_pair((WatchLink *)NULL, err);
+        return make_pair(nil, err);
 
-    WatchLink *wlink = new(WatchLink);
+    WatchLink wlink = adoptref(new(_WatchLink));
     wlink->_wc        = wc;
     wlink->_f         = f;
     wlink->_rx_eof    = makechan<structZ>();
@@ -52,8 +58,8 @@ pair<WatchLink*, error> WCFS::_openwatch() {
     return make_pair(wlink, nil);
 }
 
-void WatchLink::_closeTX() {
-    WatchLink &wlink = *this;
+void _WatchLink::_closeTX() {
+    _WatchLink &wlink = *this;
     wlink._txclose1.do_([&]() {
         // ask wcfs to close its tx & rx sides; close(wcfs.tx) wakes up
         // _serveRX on client (= on us). The connection can be already closed
@@ -69,8 +75,8 @@ void WatchLink::_closeTX() {
 }
 
 // close closes the link.
-error WatchLink::close() {
-    WatchLink& wlink = *this;
+error _WatchLink::close() {
+    _WatchLink& wlink = *this;
 
     wlink._closeTX();
 #if 0
@@ -97,8 +103,8 @@ error WatchLink::close() {
 }
 
 // _serveRX receives messages from ._f and dispatches them according to streamID.
-error WatchLink::_serveRX(context::Context ctx) {    // XXX error -> where ?
-    WatchLink& wlink = *this;
+error _WatchLink::_serveRX(context::Context ctx) {    // XXX error -> where ?
+    _WatchLink& wlink = *this;
 
     // when finishing - wakeup everyone waiting for rx
     defer([&]() {
@@ -184,16 +190,16 @@ error WatchLink::_serveRX(context::Context ctx) {    // XXX error -> where ?
 //
 // multiple _send can be called in parallel - _send serializes writes.
 // XXX +ctx?
-error WatchLink::_send(StreamID stream, const string &msg) {
-    WatchLink *wlink = this;
+error _WatchLink::_send(StreamID stream, const string &msg) {
+    _WatchLink *wlink = this;
     if (msg.find('\n') != string::npos)
         panic("msg has \\n");
     string pkt = fmt::sprintf("%lu %s\n", stream, msg.c_str());
     return wlink->_write(pkt);
 }
 
-error WatchLink::_write(const string &pkt) {
-    WatchLink *wlink = this;
+error _WatchLink::_write(const string &pkt) {
+    _WatchLink *wlink = this;
 
     wlink->_txmu.lock();
     defer([&]() {
@@ -209,8 +215,8 @@ error WatchLink::_write(const string &pkt) {
 
 // sendReq sends client -> server request and returns server reply.
 // XXX -> reply | None when EOF
-pair<string, error> WatchLink::sendReq(context::Context ctx, const string &req) {
-    WatchLink *wlink = this;
+pair<string, error> _WatchLink::sendReq(context::Context ctx, const string &req) {
+    _WatchLink *wlink = this;
     // XXX err ctx
 
     rxPkt       rx;
@@ -231,8 +237,8 @@ pair<string, error> WatchLink::sendReq(context::Context ctx, const string &req)
     return make_pair(reply, nil);
 }
 
-tuple</*rxq*/chan<rxPkt>, error> WatchLink::_sendReq(context::Context ctx, const string &req) {
-    WatchLink *wlink = this;
+tuple</*rxq*/chan<rxPkt>, error> _WatchLink::_sendReq(context::Context ctx, const string &req) {
+    _WatchLink *wlink = this;
     // XXX err ctx?
 
     wlink->_txmu.lock(); // XXX -> atomic (currently uses arbitrary lock)
@@ -269,8 +275,8 @@ tuple</*rxq*/chan<rxPkt>, error> WatchLink::_sendReq(context::Context ctx, const
 
 // recvReq receives client <- server request.
 static error _parsePinReq(PinReq *pin, const rxPkt *pkt);
-error WatchLink::recvReq(context::Context ctx, PinReq *prx) {
-    WatchLink& wlink = *this;
+error _WatchLink::recvReq(context::Context ctx, PinReq *prx) {
+    _WatchLink& wlink = *this;
     // XXX err ctx?
 
     rxPkt pkt;
@@ -329,8 +335,8 @@ static error _parsePinReq(PinReq *pin, const rxPkt *pkt) {
 }
 
 // _readline reads next raw line sent from wcfs.
-tuple<string, error> WatchLink::_readline() {
-    WatchLink& wlink = *this;
+tuple<string, error> _WatchLink::_readline() {
+    _WatchLink& wlink = *this;
     char buf[128];
 
     size_t nl_searchfrom = 0;

wcfs/internal/wcfs_watchlink.h

@@ -66,7 +66,9 @@ static_assert(sizeof(rxPkt) == 128);
 // .close() closes the link.
 //
 // It is safe to use WatchLink from multiple threads simultaneously.
-class WatchLink {
+class _WatchLink;
+typedef refptr<_WatchLink> WatchLink;
+class _WatchLink : public object {
     WCFS            *_wc;
     os::File        _f;      // head/watch file handle
     string          _rxbuf;  // buffer for read data from _f
@@ -89,8 +91,15 @@ class WatchLink {
     sync.WorkGroup *_serveWG
 #endif
 
+    // don't new - create only via WCFS._openwatch()
+private:
+    _WatchLink();
+    ~_WatchLink();
+    friend pair<WatchLink, error> WCFS::_openwatch();
+public:
+    void decref();
+
 public:
-    friend pair<WatchLink*, error> WCFS::_openwatch();
     error close();
     error recvReq(context::Context ctx, PinReq *rx_into);
     pair<string, error> sendReq(context::Context ctx, const string &req);