diff --git a/fastlib/u/garryb/nbr/build.py b/fastlib/u/garryb/nbr/build.py index fec338ab10..58773492de 100644 --- a/fastlib/u/garryb/nbr/build.py +++ b/fastlib/u/garryb/nbr/build.py @@ -1,28 +1,35 @@ librule(name = "nbr", sources = ["blockdev.cc", "nbr_utils.cc", - "cache.cc", "work.cc", "rpc_sock.cc"], + "cache.cc", "work.cc", + "rpc.cc", "rpc_sock.cc"], headers = ["blockdev.h", "cache.h", "cachearray.h", "dfs.h", "gnp.h", "kdtree.h", "nbr_utils.h", - "spbounds.h", "spnode.h", "work.h", "rpc_sock.h"], + "spbounds.h", "spnode.h", "work.h", + "rpc.cc", "rpc_sock.h"], deplibs = ["fastlib:fastlib_int"]) -librule(name = "nbr_mpi", - sources = ["rpc.cc", "netcache.cc"], - headers = ["rpc.h", "netcache.h"], +binrule(name = "rpc_sock_test", + sources = ["rpc_sock_test.cc"], deplibs = [":nbr"]) +# No more MPI +#librule(name = "nbr_mpi", +# sources = ["rpc.cc", "netcache.cc"], +# headers = ["rpc.h", "netcache.h"], +# deplibs = [":nbr"]) + binrule(name = "tkde", sources = ["tkde.cc"], deplibs = [":nbr"]) -binrule(name = "tkde_mpi", - sources = ["tkde.cc"], - deplibs = [":nbr_mpi"]) +#binrule(name = "tkde_mpi", +# sources = ["tkde.cc"], +# deplibs = [":nbr_mpi"]) -binrule(name = "allnn_mpi", - sources = ["allnn.cc"], - deplibs = [":nbr_mpi"]) +#binrule(name = "allnn_mpi", +# sources = ["allnn.cc"], +# deplibs = [":nbr_mpi"]) binrule(name = "allnn", sources = ["allnn.cc"], diff --git a/fastlib/u/garryb/nbr/nbr_utils.h b/fastlib/u/garryb/nbr/nbr_utils.h index f991014c29..66f5721722 100644 --- a/fastlib/u/garryb/nbr/nbr_utils.h +++ b/fastlib/u/garryb/nbr/nbr_utils.h @@ -270,7 +270,8 @@ void MonochromaticDualTreeMain(datanode *module, const char *gnp_name) { - write code that detects and runs the server */ -//problem problem - no network thread! +#warning "RPC is currently disabled" +#if 0 template class RpcMonochromaticDualTreeRunner { @@ -504,7 +505,9 @@ void RpcMonochromaticDualTreeMain(datanode *module, const char *gnp_name) { RpcMonochromaticDualTreeRunner runner; runner.Main(module, gnp_name); } +#endif }; + #endif diff --git a/fastlib/u/garryb/nbr/netcache.h b/fastlib/u/garryb/nbr/netcache.h index 070f0260e7..af5bf977b7 100644 --- a/fastlib/u/garryb/nbr/netcache.h +++ b/fastlib/u/garryb/nbr/netcache.h @@ -88,7 +88,7 @@ class HashedRemoteBlockDevice offset_t begin, offset_t end, const char *data); virtual blockid_t AllocBlock(); - RawRemoteObjectBackend *server() { + Channel *server() { return &server_; } int channel() const { @@ -113,7 +113,7 @@ class SimpleDistributedCacheArray : public CacheArray { public: void Configure(int channel) { - remote_device_.Init(channel, RpcImpl::rank(), RpcImpl::n_peers()); + remote_device_.Init(channel, rpc::rank(), rpc::n_peers()); } void InitMaster(const T& default_obj, @@ -141,7 +141,7 @@ class SimpleDistributedCacheArray : public CacheArray { small_cache_.Clear(mode); } - RawRemoteObjectBackend *server() { + Channel *server() { return remote_device_.server(); } diff --git a/fastlib/u/garryb/nbr/rpc.cc b/fastlib/u/garryb/nbr/rpc.cc index f83d87d78b..1c195beb7b 100644 --- a/fastlib/u/garryb/nbr/rpc.cc +++ b/fastlib/u/garryb/nbr/rpc.cc @@ -28,31 +28,33 @@ class BarrierChannel : public Channel { } void CheckState_() { - if (n_received_ >= RpcImpl::children().size()) { - if (RpcImpl::is_root() || n_received_ > RpcImpl::children().size()) { + if (n_received_ >= rpc::children().size()) { + if (rpc::is_root() || n_received_ > rpc::children().size()) { // Tell the kids that the root is ready - for (int i = 0; i < RpcImpl::children().size(); i++) { - DoMessage_(RpcImpl::children()[i]); + for (int i = 0; i < rpc::children().size(); i++) { + //fprintf(stderr, "barrier: Message to %d\n", rpc::children()[i]); + DoMessage_(rpc::children()[i]); } Done(); - RpcImpl::Unregister(channel()); + rpc::Unregister(channel()); mutex_.Lock(); done_ = true; cond_.Signal(); mutex_.Unlock(); } else { // Tell parent that all my kids are ready - DoMessage_(RpcImpl::parent()); + //fprintf(stderr, "barrier: Message to parent %d\n", rpc::parent()); + DoMessage_(rpc::parent()); } } } bool IsValidSender_(int peer) { - if (n_received_ == RpcImpl::children().size()) { - return peer == RpcImpl::parent(); + if (n_received_ == rpc::children().size()) { + return peer == rpc::parent(); } else { - for (int i = 0; i < RpcImpl::children().size(); i++) { - if (peer == RpcImpl::children()[i]) { + for (int i = 0; i < rpc::children().size(); i++) { + if (peer == rpc::children()[i]) { return true; } } @@ -62,7 +64,7 @@ class BarrierChannel : public Channel { public: BarrierTransaction() {} - ~BarrierTransaction() {} + virtual ~BarrierTransaction() {} void Doit(int channel_num) { Transaction::Init(channel_num); @@ -77,23 +79,28 @@ class BarrierChannel : public Channel { } void HandleMessage(Message *message) { - if (unlikely(!IsValidSender(message->peer()))) { + //fprintf(stderr, "barrier: Message from %d\n", message->peer()); + if (unlikely(!IsValidSender_(message->peer()))) { FATAL("Message from %d unexpected during barrier #%d with n_received=%d", message->peer(), channel(), n_received_); } delete message; n_received_++; - CheckCompletion(); + CheckState_(); } }; private: - BarrierTransaction transaction; + BarrierTransaction transaction_; public: + BarrierChannel() {} + virtual ~BarrierChannel() {} + void Doit(int channel_num) { - RpcImpl::Register(channel_num, this); - transaction.Doit(channel_num); + //fprintf(stderr, "barrier: I exist\n"); + rpc::Register(channel_num, this); + transaction_.Doit(channel_num); } Transaction *GetTransaction(Message *message) { diff --git a/fastlib/u/garryb/nbr/rpc.h b/fastlib/u/garryb/nbr/rpc.h index 1381674c39..6e6d11ee74 100644 --- a/fastlib/u/garryb/nbr/rpc.h +++ b/fastlib/u/garryb/nbr/rpc.h @@ -24,6 +24,7 @@ template class Rpc { FORBID_COPY(Rpc); private: + template struct RpcRequestTransaction : public Transaction { FORBID_COPY(RpcRequestTransaction); @@ -34,7 +35,7 @@ class Rpc { public: RpcRequestTransaction() {} - ~RpcRequestTransaction() {} + virtual ~RpcRequestTransaction() {} Message *Doit(int channel, int peer, const RequestObject& request) { Transaction::Init(channel); @@ -54,8 +55,8 @@ class Rpc { void HandleMessage(Message *message) { mutex.Lock(); response = message; - mutex.Unlock(); cond.Signal(); + mutex.Unlock(); // TODO: Handle done Done(); } @@ -79,9 +80,9 @@ class Rpc { template ResponseObject *Request( int channel, int peer, const RequestObject& request) { - RpcRequestTransaction transaction; + RpcRequestTransaction transaction; response_ = transaction.Doit(channel, peer, request); - response_object_ = ot::PointerThaw(response_.data()); + response_object_ = ot::PointerThaw(response_->data()); return response_object_; } @@ -95,7 +96,7 @@ class Rpc { return *response_object_; } operator const ResponseObject *() const { - return response_objet_; + return response_object_; } const ResponseObject* operator ->() const { return response_object_; @@ -112,7 +113,7 @@ template class RemoteObjectBackend : public Channel { public: // Simple request-response transaction - class RemoteObjectTransaction() { + class RemoteObjectTransaction : public Transaction { FORBID_COPY(RemoteObjectTransaction); private: RemoteObjectBackend *inner_; @@ -123,7 +124,7 @@ class RemoteObjectBackend : public Channel { {} void HandleMessage(Message *request); - } + }; public: virtual ~RemoteObjectBackend() {} @@ -131,7 +132,7 @@ class RemoteObjectBackend : public Channel { virtual void HandleRequest(const RequestObject& request, ResponseObject *response) = 0; - RemoteObjectTransaction *GetTransaction(Message *message); { + RemoteObjectTransaction *GetTransaction(Message *message) { return new RemoteObjectTransaction(this); } diff --git a/fastlib/u/garryb/nbr/rpc_sock.cc b/fastlib/u/garryb/nbr/rpc_sock.cc index e573ac9494..bc249cc555 100644 --- a/fastlib/u/garryb/nbr/rpc_sock.cc +++ b/fastlib/u/garryb/nbr/rpc_sock.cc @@ -4,6 +4,15 @@ * Implementation of transaction API using TCP. */ +#include "rpc.h" +#include "rpc_sock.h" + +#include "fastlib/fastlib.h" + +#include +#include +#include + /* tasks to complete that must work - startup and shutdown @@ -49,9 +58,9 @@ RpcSockImpl RpcSockImpl::instance; void RpcSockImpl::Init() { module_ = fx_submodule(fx_root, "rpc", "rpc"); - rank_ = fx_param_int(module_, "rank", "0"); n_peers_ = fx_param_int_req(module_, "n"); - port_ = fax_param_int(module_, "port", 31415); + rank_ = fx_param_int(module_, "rank", 0); + port_ = fx_param_int(module_, "port", 31415); channels_.Init(); channels_.default_value() = NULL; @@ -60,14 +69,16 @@ void RpcSockImpl::Init() { Listen_(); StartPollingThread_(); - fprintf("%d: Starting initial barrier\n", rank_); + //fprintf(stderr, "%d: Starting initial barrier\n", rank_); rpc::Barrier(0); - fprintf("%d: Initial barrier over!\n", rank_); + //fprintf(stderr, "%d: Initial barrier over!\n", rank_); } void RpcSockImpl::Done() { status_ = STOP_SYNC; + //fprintf(stderr, "%d: Starting final barrier\n", rank_); rpc::Barrier(1); + //fprintf(stderr, "%d: Finished final barrier!\n", rank_); status_ = STOP; polling_thread_.WaitStop(); close(listen_fd_); @@ -75,7 +86,7 @@ void RpcSockImpl::Done() { } void RpcSockImpl::Register(int channel_num, Channel *channel) { - mutex_.Lock() + mutex_.Lock(); channels_[channel_num] = channel; mutex_.Unlock(); // Inform the polling loop about the new channel so that it can process @@ -84,7 +95,7 @@ void RpcSockImpl::Register(int channel_num, Channel *channel) { } void RpcSockImpl::Unregister(int channel_num) { - mutex_.Lock() + mutex_.Lock(); channels_[channel_num] = NULL; mutex_.Unlock(); } @@ -100,21 +111,20 @@ void RpcSockImpl::WakeUpPollingLoop() { (void) write(alert_signal_fd_, "x", 1); } -int RpcSockImpl::UnregisterTransaction(int peer_id, int channel, int id) { +void RpcSockImpl::UnregisterTransaction(int peer_id, int channel, int id) { Peer *peer = &peers_[peer_id]; peer->mutex.Lock(); // Lock peer's mutex if (channel < 0) { peer->outgoing_transactions[id] = NULL; } else { - mutex_.Lock(); // Lock mutex -- we are accessing channels - channels_[channel]->CleanupTransaction(incoming_transactions[id]); - mutex_.Unlock(); + // Old idea -- not really necessary + //mutex_.Lock(); // Lock mutex -- we are accessing channels + //channels_[channel]->CleanupTransaction(incoming_transactions[id]); + //mutex_.Unlock(); peer->incoming_transactions[id] = NULL; } peer->mutex.Unlock(); - - return id; } int RpcSockImpl::AssignTransaction(int peer_num, Transaction *transaction) { @@ -141,7 +151,7 @@ void RpcSockImpl::CreatePeers_() { for (index_t i = 0; i < peers_.size(); i++) { Peer *peer = &peers_[i]; - peer->connection.Init(reader.Peek().c_str(), port_); + peer->connection.Init(i, reader.Peek().c_str(), port_); reader.Gobble(); } } @@ -151,14 +161,16 @@ void RpcSockImpl::CalcChildren_() { + min(unsigned(n_peers_ - rank_ - 1), unsigned((~rank_) & (rank_-1))); int i; + children_.Init(); for (i = 1; i < m; i *= 2) {} while (i > 1) { i /= 2; *children_.AddBack() = rank_ + i; - DEBUG_MSG(1.0, "%d child: %d\n", rank_, rank_ + i); + //fprintf(stderr, "children: %d child: %d\n", rank_, rank_ + i); } parent_ = rank_ - ((~rank_) & (rank_-1)) - 1; + //fprintf(stderr, "parent = %d\n", parent_); } void RpcSockImpl::Listen_() { @@ -168,17 +180,18 @@ void RpcSockImpl::Listen_() { socketpair(AF_LOCAL, SOCK_STREAM, 0, sv); alert_signal_fd_ = sv[0]; alert_slot_fd_ = sv[1]; - MakeSocketNonblocking(alert_signal_fd_); - MakeSocketNonblocking(alert_slot_fd_); + MakeSocketNonBlocking(alert_signal_fd_); + MakeSocketNonBlocking(alert_slot_fd_); - listen_fd_ = socket(AF_INET, SOCK_STREAM, PF_INET); // last param 0? + listen_fd_ = socket(AF_INET, SOCK_STREAM, 0); // last param 0? mem::Zero(&my_address); my_address.sin_family = AF_INET; my_address.sin_port = htons(port_); my_address.sin_addr.s_addr = htonl(INADDR_ANY); if (0 > bind(listen_fd_, (struct sockaddr*)&my_address, sizeof(my_address))) { - FATAL("Could not bind to selected port %d", port_); + FATAL("Could not bind to selected port %d on fd %d: %s", + port_, listen_fd_, strerror(errno)); } if (0 > listen(listen_fd_, 10)) { @@ -210,8 +223,8 @@ void RpcSockImpl::PollingLoop_() { FD_ZERO(&write_fds); FD_ZERO(&error_fds); - FD_SET(&read_fds, listen_fd_); - FD_SET(&read_fds, alert_slot_fd_); + FD_SET(listen_fd_, &read_fds); + FD_SET(alert_slot_fd_, &read_fds); maxfd = max(listen_fd_, alert_slot_fd_); for (index_t i = 0; i < peers_.size(); i++) { @@ -227,11 +240,17 @@ void RpcSockImpl::PollingLoop_() { // Use a one-second timeout so we can poll for should_stop_ to allow // graceful shutdown. - fprintf("%d: select()", rank_); struct timeval tv; - tv.tv_sec = 2; + tv.tv_sec = 1; tv.tv_usec = 0; - int n_events = select(maxfd + 1, &read_fds, &write_fds, &error_fds, &ts); + int n_events = select(maxfd + 1, &read_fds, &write_fds, &error_fds, &tv); + //fprintf(stderr, "%d: select() returns %d\n", rank_, n_events); + + //for (int i = 0; i < maxfd; i++) { + // if (FD_ISSET(i, &read_fds)) { //fprintf(stderr, "%d: read on %d\n", rank_, i); } + // if (FD_ISSET(i, &write_fds)) { //fprintf(stderr, "%d: write on %d\n", rank_, i); } + // if (FD_ISSET(i, &error_fds)) { //fprintf(stderr, "%d: error on %d\n", rank_, i); } + //} if (n_events < 0) { NONFATAL("Select failed"); @@ -242,14 +261,15 @@ void RpcSockImpl::PollingLoop_() { continue; } - if (FD_ISSET(&read_fds, alert_signal_fd_)) { + if (FD_ISSET(alert_slot_fd_, &read_fds)) { // We got a wake-up signal. Clear the buffer so we won't receive the // signal twice. char buf[8]; - while (read(alert_signal_fd_, buf, sizeof(buf)) > 0) {} + while (read(alert_slot_fd_, buf, sizeof(buf)) > 0) {} } - if (FD_ISSET(&read_fds, listen_fd_)) { + if (FD_ISSET(listen_fd_, &read_fds)) { + //fprintf(stderr, "%d: Connection.\n", rank_); // Accept incoming connections for (;;) { sockaddr_in addr; @@ -263,12 +283,18 @@ void RpcSockImpl::PollingLoop_() { break; } - int i; + index_t i; for (i = 0; i < peers_.size(); i++) { Peer *peer = &peers_[i]; + // We don't have to lock here since sin_addr is not locked + // TODO: Consider moving the locks elsewhere if (addr.sin_addr.s_addr == peer->connection.peer_addr().sin_addr.s_addr) { + peer->mutex.Lock(); + peer->connection.AcceptIncoming(new_fd); + peer->mutex.Unlock(); + //fprintf(stderr, "%d: accepted rank %d on fd %d\n", rank_, i, new_fd); break; } } @@ -276,11 +302,6 @@ void RpcSockImpl::PollingLoop_() { NONFATAL("Incomming connection from unknown machine, %s.", inet_ntoa(addr.sin_addr)); (void)close(new_fd); - } else { - DEBUG_ASSERT(peers_[i].incoming_connection == NULL); - SockConnection *connection = new SockConnection(); - connection->Init(new_fd); - peers_[i].incoming_connection = connection; } } } @@ -296,7 +317,7 @@ void RpcSockImpl::PollingLoop_() { // especially not having to lock all the mutexes. // How about a DenseIntMap? Or, we can have the mutex be part of the // connection itself? - peer->mutex.Lock() + peer->mutex.Lock(); // we'll allow errors to occur if we're shutting down. peer->connection.HandleSocketEvents(&read_fds, &write_fds, &error_fds, status_ != RUN); @@ -315,11 +336,12 @@ void RpcSockImpl::PollingLoop_() { } void RpcSockImpl::GatherReadyMessages_(Peer *peer, - ArrayList *work_items) { + ArrayList *work_items) { ArrayList* queue = &peer->connection.read_queue(); - int j = 0; + index_t j = 0; + index_t i; - for (index_t i = 0; i < queue->size(); i++) { + for (i = 0; i < queue->size(); i++) { Message *message = (*queue)[i]; int id = message->transaction_id(); Transaction *transaction; @@ -337,7 +359,7 @@ void RpcSockImpl::GatherReadyMessages_(Peer *peer, Channel *channel = channels_[message->channel()]; if (channel) { transaction = channel->GetTransaction(message); - item->transaction->TransactionHandleNewSender_(item->message); + transaction->TransactionHandleNewSender_(message); peer->incoming_transactions[id] = transaction; } } @@ -347,7 +369,7 @@ void RpcSockImpl::GatherReadyMessages_(Peer *peer, // This work item is processable, add it to the transactions WorkItem *item = work_items->AddBack(); item->message = message; - item->transcation = transaction; + item->transaction = transaction; } else { // No good... we have to enqueue it. (*queue)[j++] = message; @@ -397,7 +419,7 @@ Message *Transaction::CreateMessage(int peer, size_t size) { peers_.AddBack(); transaction_id = RpcSockImpl::instance.AssignTransaction(peer, this); peers_[i].peer = peer; - peers_[i].channel = channel; + peers_[i].channel = channel(); peers_[i].transaction_id = transaction_id; } @@ -414,22 +436,23 @@ void Transaction::TransactionHandleNewSender_(Message *message) { // We'll reply to this with channel -1, meaning that it was the other end // who initiated the transaction ID, i.e., the transaction ID lives in // their namespace. - Peer *peer = peers_.AddBack(); - peer->peer = message->peer(); - peer->channel = -1; - peer->transaction_id = message->transaction_id(); + PeerInfo *peer_info = peers_.AddBack(); + peer_info->peer = message->peer(); + peer_info->channel = -1; + peer_info->transaction_id = message->transaction_id(); } void Transaction::Send(Message *message) { // RpcSockImpl knows how to send messages, we don't need to bother with it. - RpcSockImpl::Send(message); + RpcSockImpl::instance.Send(message); } void Transaction::Done() { for (index_t i = 0; i < peers_.size(); i++) { RpcSockImpl::instance.UnregisterTransaction( - peers[i].peer, peers[i].channel, peers[i].transaction_id); + peers_[i].peer, peers_[i].channel, peers_[i].transaction_id); } + peers_.Clear(); } void Transaction::Done(int peer) { @@ -437,18 +460,24 @@ void Transaction::Done(int peer) { // TODO: Demeter? if (peer == peers_[i].peer) { RpcSockImpl::instance.UnregisterTransaction( - peers[i].peer, peers[i].channel, peers[i].transaction_id); + peers_[i].peer, peers_[i].channel, peers_[i].transaction_id); peers_[i] = peers_[peers_.size()-1]; peers_.PopBack(); + break; } } } //------------------------------------------------------------------------- -~SockConnection() { +SockConnection::~SockConnection() { // TODO: There are more socket functions I might have to call - (void) close(fd_); + if (is_read_open()) { + (void) close(read_fd_); + } + if (is_write_open()) { + (void) close(write_fd_); + } } Message *SockConnection::CreateMessage( @@ -467,8 +496,11 @@ Message *SockConnection::CreateMessage( return message; } -void SockConnection::Init(const char *ip_address, int port) { +void SockConnection::Init(int peer_num, const char *ip_address, int port) { + peer_ = peer_num; + mem::Zero(&peer_addr_); + peer_addr_.sin_family = AF_INET; peer_addr_.sin_port = htons(port); if (inet_pton(AF_INET, ip_address, &peer_addr_.sin_addr) < 0) { FATAL("Invalid IP address [%s] -- must be 1.2.3.4 format\n", ip_address); @@ -489,12 +521,14 @@ void SockConnection::Init(const char *ip_address, int port) { } void SockConnection::OpenOutgoing() { - outgoing_fd_ = socket(AF_INET, SOCK_STREAM, PF_INET); - MakeSocketNonBlocking(outgoing_fd_); + write_fd_ = socket(AF_INET, SOCK_STREAM, 0); - if (0 > connect(fd, &peer_addr_, sizeof(struct sockaddr_in)) + MakeSocketNonBlocking(write_fd_); + + //fprintf(stderr, "connect to peer %d, %s\n", peer_, inet_ntoa(peer_addr_.sin_addr)); + if (0 > connect(write_fd_, (struct sockaddr*)&peer_addr_, sizeof(struct sockaddr_in)) && errno != EINTR && errno != EINPROGRESS) { - FATAL("connect failed"); + FATAL("connect failed: %s", strerror(errno)); } // We'd nominally have to wake up the polling loop here to inform the @@ -503,9 +537,9 @@ void SockConnection::OpenOutgoing() { // NULL. } -void SockConnection::AcceptIncoming(int fd) { - incoming_fd_ = fd; - MakeSocketNonBlocking(incoming_fd_); +void SockConnection::AcceptIncoming(int accepted_fd) { + read_fd_ = accepted_fd; + MakeSocketNonBlocking(read_fd_); } void SockConnection::Send(Message *message) { @@ -514,7 +548,7 @@ void SockConnection::Send(Message *message) { OpenOutgoing(); } - ++write_total; + ++write_total_; if (likely(write_message_ == NULL)) { // If we're not writing anything now, set our current message. write_buffer_pos_ = 0; @@ -527,12 +561,22 @@ void SockConnection::Send(Message *message) { } void SockConnection::TryWrite() { - // First, see if we're *trying* to write something. - if (is_writing()) { + while (is_writing()) { + if (write_buffer_pos_ == write_message_->buffer_size()) { + // Looks like we successfully wrote the whole message. + delete write_message_; + if (write_queue_.is_empty()) { + write_message_ = NULL; + break; + } else { + write_buffer_pos_ = 0; + write_message_ = write_queue_.Pop(); + } + } // Try to write something. ssize_t bytes_written = write(write_fd_, - write_message_.buffer() + write_buffer_pos_, - write_message_.buffer_size() - write_buffer_pos_); + write_message_->buffer() + write_buffer_pos_, + write_message_->buffer_size() - write_buffer_pos_); if (bytes_written < 0) { // Okay, we weren't able to write anything. if (errno != EAGAIN && errno != EINTR) { @@ -543,66 +587,68 @@ void SockConnection::TryWrite() { } else { // We successfully wrote something, update our position. write_buffer_pos_ += bytes_written; - if (write_buffer_pos_ == write_message_.buffer_size()) { - // Looks like we successfully wrote the whole message. - delete write_message_; - if (write_queue_.is_empty()) { - write_message_ = NULL; - } else { - write_buffer_pos_ = 0; - write_message_ = write_queue_.Pop(); - } - } } } } void SockConnection::TryRead() { - if (!read_message_) { - // It looks like we weren't in the process of reading another packet, so - // this is a new packet. - Header header; - // Read the packet's header. - ssize_t header_bytes = read(read_fd_, - reinterpret_cast(&header), sizeof(Header)); - if (header_bytes < 0 && (errno == EINTR || errno == EAGAIN)) { - return; + //fprintf(stderr, "Trying to read!\n"); + for (;;) { + // First, read a header if we have to. + if (!read_message_) { + // It looks like we weren't in the process of reading another packet, so + // this is a new packet. + Header header; + // Read the packet's header. + ssize_t header_bytes = read(read_fd_, &header, sizeof(Header)); + // Error out if we got the wrong number of bytes. + // In theory, we probably can't always assume the headers won't be be + // chopped up. + if (header_bytes != sizeof(Header)) { + if (header_bytes == 0 || errno == EINTR || errno == EAGAIN) { + // okay, it looks like there isn't any data + //fprintf(stderr, "Looks like we don't actually have data...\n"); + return; + } else { + FATAL("Error reading packet header: read returned %d bytes: %s", + int(header_bytes), strerror(errno)); + } + } + DEBUG_ASSERT(header.magic == MAGIC); + // When we read in a message, we don't need to allocate space for the + // header (since we have already read it successfully). + read_message_ = new Message(); + read_message_->Init(peer_, header.channel, header.transaction_id, + mem::Alloc(header.data_size), 0, header.data_size); + read_buffer_pos_ = 0; + //fprintf(stderr, "Got a valid header.\n"); } - // Error out if we got the wrong number of bytes. - // In theory, we probably can't always assume the headers won't be be - // chopped up. - if (header_bytes != sizeof(Header)) { - FATAL("Error reading packet header: read returned %d bytes", - int(header_bytes)); - } - DEBUG_ASSERT(header.magic == MAGIC); - // When we read in a message, we don't need to allocate space for the - // header (since we have already read it successfully). - read_message_ = new Message(); - read_message_->Init(peer_, header.channel, header.transaction_id, - mem::Alloc(header.data_size), 0, header.data_size); - read_buffer_pos_ = 0; - } - ssize_t bytes_read = read(read_fd_, - read_message_.data() + read_buffer_pos_, - read_message_.data_size() - read_buffer_pos_); - if (bytes_read < 0) { - // Couldn't read anything. - if (errno != EAGAIN && errno != EINTR) { - // Error wasn't due to the fact that it's non-blocking, so the socket - // was disconnected. - FATAL("Error reading"); - } - } else { - // Progress our position in the buffer. - read_buffer_pos_ += bytes_read; - if (read_buffer_pos_ == read_message_.buffer_size()) { + // Second, see if we're done with the packet. (Note some packets have + // a null message length!) + if (read_buffer_pos_ == read_message_->buffer_size()) { // We've read a whole message. Put it on the queue to be serviced. - ++read_total; - *read_queue_.AddBack() = message; + ++read_total_; + *read_queue_.AddBack() = read_message_; read_message_ = NULL; read_buffer_pos_ = 0; + break; + } + // Finally, read as much payload as we can for this message. + ssize_t bytes_read = read(read_fd_, + read_message_->data() + read_buffer_pos_, + read_message_->data_size() - read_buffer_pos_); + //fprintf(stderr, "Got %d data bytes.\n", (int)bytes_read); + if (bytes_read > 0) { + read_buffer_pos_ += bytes_read; + } else { + // Couldn't read anything. + if (bytes_read != 0 && errno != EAGAIN && errno != EINTR) { + // Error wasn't due to the fact that it's non-blocking, so the socket + // was disconnected. + FATAL("Error reading"); + } + break; } } } @@ -618,40 +664,44 @@ void SockConnection::HandleSocketEvents( &sockError, &sockErrorLen) == -1) { ... */ - if (FD_ISSET(&error_fds, read_fd_)) { - // Poor man's way to terminate all processes - if (allow_errors) { - read_fd_ = -1; - } else { - FATAL("Socket error on read fd"); + if (unlikely(is_read_open())) { + if (FD_ISSET(read_fd_, error_fds)) { + // Poor man's way to terminate all processes + if (allow_errors) { + read_fd_ = -1; + } else { + FATAL("Socket error on read fd"); + } + } + if (FD_ISSET(read_fd_, read_fds)) { + TryRead(); } } - if (FD_ISSET(&error_fds, write_fd_)) { - // Poor man's way to terminate all processes - if (allow_errors) { - write_fd_ = -1; - } else { - FATAL("Socket error on out fd"); + if (unlikely(is_write_open())) { + if (FD_ISSET(write_fd_, error_fds)) { + // Poor man's way to terminate all processes + if (allow_errors) { + write_fd_ = -1; + } else { + FATAL("Socket error on out fd"); + } + } + if (FD_ISSET(write_fd_, write_fds)) { + TryWrite(); } - } - if (FD_ISSET(&write_fds, write_fd_)) { - TryWrite(); - } - if (FD_ISSET(&read_fds, read_fd_)) { - TryRead(); } } void SockConnection::PrepareSelect( fd_set *read_fds, fd_set *write_fds, fd_set *error_fds) { if (is_read_open()) { - FD_SET(read_fds, read_fd_); - FD_SET(error_fds, read_fd_); + FD_SET(read_fd_, read_fds); + FD_SET(read_fd_, error_fds); } if (is_write_open()) { if (is_writing()) { - FD_SET(write_fds, write_fd_); + FD_SET(write_fd_, write_fds); } - FD_SET(error_fds, write_fd_); + FD_SET(write_fd_, error_fds); } } diff --git a/fastlib/u/garryb/nbr/rpc_sock.h b/fastlib/u/garryb/nbr/rpc_sock.h index 872fd72521..c91c6c11f8 100644 --- a/fastlib/u/garryb/nbr/rpc_sock.h +++ b/fastlib/u/garryb/nbr/rpc_sock.h @@ -7,6 +7,8 @@ #ifndef RPC_SOCK_H #define RPC_SOCK_H +#include "fastlib/fastlib_int.h" + #include #include #include @@ -56,7 +58,7 @@ class Message { return buffer_; } int peer() const { - return peer_in_; + return peer_; } int channel() const { return channel_; @@ -67,6 +69,8 @@ class Message { }; class Transaction { + FORBID_COPY(Transaction); + private: int channel_; struct PeerInfo { @@ -76,10 +80,10 @@ class Transaction { }; ArrayList peers_; - public: - static Message *CreateMessage(int peer, size_t size); - protected: + /** Create a message of a specified size, which you will later Send(). */ + Message *CreateMessage(int peer, size_t size); + /** Send a message */ void Send(Message *message); /** Unregister the transaction from all peers */ void Done(); @@ -87,6 +91,9 @@ class Transaction { void Done(int peer); public: + Transaction() {} + virtual ~Transaction() {} + void Init(int channel_num); int channel() const { @@ -140,17 +147,17 @@ class SockConnection { int peer, int channel, int transaction_id, size_t size); private: - WALDO + int peer_; int read_fd_; int write_fd_; struct sockaddr_in peer_addr_; - int read_total; + int read_total_; Message *read_message_; size_t read_buffer_pos_; ArrayList read_queue_; - int write_total; + int write_total_; Message *write_message_; size_t write_buffer_pos_; MinHeap write_queue_; @@ -160,7 +167,7 @@ class SockConnection { ~SockConnection(); /** Creates an unopened SocketConnection placeholder. */ - void Init(const char *ip_address, int port); + void Init(int peer, const char *ip_address, int port); /** Create an outgoing connection for sending messages. */ void OpenOutgoing(); /** Accept an incoming connection for receiving messages. */ @@ -260,7 +267,7 @@ class RpcSockImpl { /** * Task structure just so we can run the polling loop in another thread. */ - class PollingTask : Task { + class PollingTask : public Task { private: RpcSockImpl *main_object_; public: @@ -333,7 +340,7 @@ class RpcSockImpl { void Listen_(); void StartPollingThread_(); void PollingLoop_(); - void GatherReadyMessages_(Peer *peer, ArrayList* work_items); + void GatherReadyMessages_(Peer *peer, ArrayList* work_items); }; namespace rpc { diff --git a/fastlib/u/garryb/nbr/rpc_sock_test.cc b/fastlib/u/garryb/nbr/rpc_sock_test.cc index 4da4ddf145..397e36f839 100644 --- a/fastlib/u/garryb/nbr/rpc_sock_test.cc +++ b/fastlib/u/garryb/nbr/rpc_sock_test.cc @@ -3,7 +3,9 @@ int main(int argc, char *argv[]) { fx_init(argc, argv); rpc::Init(); + fprintf(stderr, "Initialized.\n"); rpc::Done(); + fprintf(stderr, "Done.\n"); fx_done(); }