Got rid of most using directives. Ported tests to python.

[invirt/third/libt4.git] / rpc / connection.cc

diff --git a/rpc/connection.cc b/rpc/connection.cc
index d118539..c7e8f95 100644 (file)
--- a/rpc/connection.cc
+++ b/rpc/connection.cc
@@ -30,7 +30,7 @@ connection::~connection() {
     // will be active
     poll_mgr::shared_mgr.block_remove_fd(fd);
     VERIFY(dead_);
     // will be active
     poll_mgr::shared_mgr.block_remove_fd(fd);
     VERIFY(dead_);

-    VERIFY(!wpdu_.buf.size());
+    VERIFY(wpdu_.status == unused);

 }
 
 shared_ptr<connection> connection::to_dst(const sockaddr_in & dst, connection_delegate * delegate, int lossy) {
 }
 
 shared_ptr<connection> connection::to_dst(const sockaddr_in & dst, connection_delegate * delegate, int lossy) {

@@ -42,28 +42,23 @@ shared_ptr<connection> connection::to_dst(const sockaddr_in & dst, connection_de
         return nullptr;
     }
     IF_LEVEL(2) LOG_NONMEMBER << "connection::to_dst fd=" << s << " to dst " << inet_ntoa(dst.sin_addr) << ":" << ntoh(dst.sin_port);
         return nullptr;
     }
     IF_LEVEL(2) LOG_NONMEMBER << "connection::to_dst fd=" << s << " to dst " << inet_ntoa(dst.sin_addr) << ":" << ntoh(dst.sin_port);

-    return make_shared<connection>(delegate, std::move(s), lossy);
+    return std::make_shared<connection>(delegate, std::move(s), lossy);

 }
 
 bool connection::send(const string & b) {
     lock ml(m_);
 
 }
 
 bool connection::send(const string & b) {
     lock ml(m_);
 

-    waiters_++;
-    while (!dead_ && wpdu_.buf.size())
+    while (!dead_ && wpdu_.status != unused)

         send_wait_.wait(ml);
         send_wait_.wait(ml);

-    waiters_--;

 
     if (dead_)
         return false;
 
 
     if (dead_)
         return false;
 

-    wpdu_.buf = b;
-    wpdu_.solong = 0;
+    wpdu_ = {inflight, b, 0};

 
 

-    if (lossy_) {
-        if ((random()%100) < lossy_) {
-            IF_LEVEL(1) LOG << "send LOSSY TEST shutdown fd " << fd;
-            shutdown(fd,SHUT_RDWR);
-        }
+    if (lossy_ && (random()%100) < lossy_) {
+        IF_LEVEL(1) LOG << "send LOSSY TEST shutdown fd " << fd;
+        shutdown(fd,SHUT_RDWR);

     }
 
     if (!writepdu()) {
     }
 
     if (!writepdu()) {

@@ -71,17 +66,15 @@ bool connection::send(const string & b) {
         ml.unlock();
         poll_mgr::shared_mgr.block_remove_fd(fd);
         ml.lock();
         ml.unlock();
         poll_mgr::shared_mgr.block_remove_fd(fd);
         ml.lock();

-    } else if (wpdu_.solong != wpdu_.buf.size()) {
+    } else if (wpdu_.status == inflight && wpdu_.cursor < b.size()) {

         // should be rare to need to explicitly add write callback
         poll_mgr::shared_mgr.add_callback(fd, CB_WRONLY, this);
         // should be rare to need to explicitly add write callback
         poll_mgr::shared_mgr.add_callback(fd, CB_WRONLY, this);

-        while (!dead_ && wpdu_.solong != size_t_max && wpdu_.solong < wpdu_.buf.size())
+        while (!dead_ && wpdu_.status == inflight && wpdu_.cursor < b.size())

             send_complete_.wait(ml);
     }
             send_complete_.wait(ml);
     }

-    bool ret = (!dead_ && wpdu_.solong == wpdu_.buf.size());
-    wpdu_.solong = 0;
-    wpdu_.buf.clear();
-    if (waiters_ > 0)
-        send_wait_.notify_all();
+    bool ret = (!dead_ && wpdu_.status == inflight && wpdu_.cursor == b.size());
+    wpdu_ = {unused, "", 0};
+    send_wait_.notify_all();

     return ret;
 }
 
     return ret;
 }
 

@@ -90,7 +83,7 @@ void connection::write_cb(int s) {
     lock ml(m_);
     VERIFY(!dead_);
     VERIFY(fd == s);
     lock ml(m_);
     VERIFY(!dead_);
     VERIFY(fd == s);

-    if (wpdu_.buf.size() == 0) {
+    if (wpdu_.status != inflight) {

         poll_mgr::shared_mgr.del_callback(fd, CB_WRONLY);
         return;
     }
         poll_mgr::shared_mgr.del_callback(fd, CB_WRONLY);
         return;
     }

@@ -98,14 +91,30 @@ void connection::write_cb(int s) {
         poll_mgr::shared_mgr.del_callback(fd, CB_RDWR);
         dead_ = true;
     } else {
         poll_mgr::shared_mgr.del_callback(fd, CB_RDWR);
         dead_ = true;
     } else {

-        VERIFY(wpdu_.solong != size_t_max);
-        if (wpdu_.solong < wpdu_.buf.size()) {
+        VERIFY(wpdu_.status != error);
+        if (wpdu_.cursor < wpdu_.buf.size())

             return;
             return;

-        }

     }
     send_complete_.notify_one();
 }
 
     }
     send_complete_.notify_one();
 }
 

+bool connection::writepdu() {
+    VERIFY(wpdu_.status == inflight);
+    if (wpdu_.cursor == wpdu_.buf.size())
+        return true;
+
+    ssize_t n = write(fd, &wpdu_.buf[wpdu_.cursor], (wpdu_.buf.size()-wpdu_.cursor));
+    if (n < 0) {
+        if (errno != EAGAIN) {
+            IF_LEVEL(1) LOG << "writepdu fd " << fd << " failure errno=" << errno;
+            wpdu_ = {error, "", 0};
+        }
+        return (errno == EAGAIN);
+    }
+    wpdu_.cursor += (size_t)n;
+    return true;
+}
+

 // fd is ready to be read
 void connection::read_cb(int s) {
     lock ml(m_);
 // fd is ready to be read
 void connection::read_cb(int s) {
     lock ml(m_);

@@ -115,7 +124,7 @@ void connection::read_cb(int s) {
 
     IF_LEVEL(5) LOG << "got data on fd " << s;
 
 
     IF_LEVEL(5) LOG << "got data on fd " << s;
 

-    if (!rpdu_.buf.size() || rpdu_.solong < rpdu_.buf.size()) {
+    if (rpdu_.status == unused || rpdu_.cursor < rpdu_.buf.size()) {

         if (!readpdu()) {
             IF_LEVEL(5) LOG << "readpdu on fd " << s << " failed; dying";
             poll_mgr::shared_mgr.del_callback(fd, CB_RDWR);
         if (!readpdu()) {
             IF_LEVEL(5) LOG << "readpdu on fd " << s << " failed; dying";
             poll_mgr::shared_mgr.del_callback(fd, CB_RDWR);

@@ -124,36 +133,17 @@ void connection::read_cb(int s) {
         }
     }
 
         }
     }
 

-    if (rpdu_.buf.size() && rpdu_.buf.size() == rpdu_.solong) {
+    if (rpdu_.status == inflight && rpdu_.buf.size() == rpdu_.cursor) {

         if (delegate_->got_pdu(shared_from_this(), rpdu_.buf)) {
             // connection_delegate has successfully consumed the pdu
         if (delegate_->got_pdu(shared_from_this(), rpdu_.buf)) {
             // connection_delegate has successfully consumed the pdu

-            rpdu_.buf.clear();
-            rpdu_.solong = 0;
+            rpdu_ = {unused, "", 0};

         }
     }
 }
 
         }
     }
 }
 

-bool connection::writepdu() {
-    VERIFY(wpdu_.solong != size_t_max);
-    if (wpdu_.solong == wpdu_.buf.size())
-        return true;
-
-    ssize_t n = write(fd, &wpdu_.buf[wpdu_.solong], (wpdu_.buf.size()-wpdu_.solong));
-    if (n < 0) {
-        if (errno != EAGAIN) {
-            IF_LEVEL(1) LOG << "writepdu fd " << fd << " failure errno=" << errno;
-            wpdu_.solong = size_t_max;
-            wpdu_.buf.clear();
-        }
-        return (errno == EAGAIN);
-    }
-    wpdu_.solong += (size_t)n;
-    return true;
-}
-

 bool connection::readpdu() {
     IF_LEVEL(5) LOG << "the receive buffer has length " << rpdu_.buf.size();
 bool connection::readpdu() {
     IF_LEVEL(5) LOG << "the receive buffer has length " << rpdu_.buf.size();

-    if (!rpdu_.buf.size()) {
+    if (rpdu_.status == unused) {

         rpc_protocol::rpc_sz_t sz1;
         ssize_t n = fd.read(sz1);
 
         rpc_protocol::rpc_sz_t sz1;
         ssize_t n = fd.read(sz1);
 

@@ -179,22 +169,20 @@ bool connection::readpdu() {
 
         IF_LEVEL(5) LOG << "read size of datagram = " << sz;
 
 
         IF_LEVEL(5) LOG << "read size of datagram = " << sz;
 

-        rpdu_.buf.assign(sz+sizeof(sz1), 0);
-        rpdu_.solong = sizeof(sz1);
+        rpdu_ = {inflight, string(sz+sizeof(sz1), 0), sizeof(sz1)};

     }
 
     }
 

-    ssize_t n = fd.read(&rpdu_.buf[rpdu_.solong], rpdu_.buf.size() - rpdu_.solong);
+    ssize_t n = fd.read(&rpdu_.buf[rpdu_.cursor], rpdu_.buf.size() - rpdu_.cursor);

 
     IF_LEVEL(5) LOG << "read " << n << " bytes";
 
     if (n <= 0) {
         if (errno == EAGAIN)
             return true;
 
     IF_LEVEL(5) LOG << "read " << n << " bytes";
 
     if (n <= 0) {
         if (errno == EAGAIN)
             return true;

-        rpdu_.buf.clear();
-        rpdu_.solong = 0;
+        rpdu_ = {unused, "", 0};

         return false;
     }
         return false;
     }

-    rpdu_.solong += (size_t)n;
+    rpdu_.cursor += (size_t)n;

     return true;
 }
 
     return true;
 }
 

@@ -239,11 +227,10 @@ void connection_listener::read_cb(int) {
     int s1 = accept(tcp_, (sockaddr *)&sin, &slen);
     if (s1 < 0) {
         perror("connection_listener::accept_conn error");
     int s1 = accept(tcp_, (sockaddr *)&sin, &slen);
     if (s1 < 0) {
         perror("connection_listener::accept_conn error");

-        throw runtime_error("connection listener failure");
+        throw std::runtime_error("connection listener failure");

     }
 
     IF_LEVEL(2) LOG << "accept_loop got connection fd=" << s1 << " " << inet_ntoa(sin.sin_addr) << ":" << ntoh(sin.sin_port);
     }
 
     IF_LEVEL(2) LOG << "accept_loop got connection fd=" << s1 << " " << inet_ntoa(sin.sin_addr) << ":" << ntoh(sin.sin_port);

-    auto ch = make_shared<connection>(delegate_, s1, lossy_);

 
     // garbage collect dead connections
     for (auto i = conns_.begin(); i != conns_.end();) {
 
     // garbage collect dead connections
     for (auto i = conns_.begin(); i != conns_.end();) {

@@ -253,5 +240,5 @@ void connection_listener::read_cb(int) {
             ++i;
     }
 
             ++i;
     }
 

-    conns_[s1] = ch;
+    conns_[s1] = std::make_shared<connection>(delegate_, s1, lossy_);

 }
 }