apaam/mlpack
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

hi

Browse Source
This commit is contained in:
Garry Boyer
2007-07-10 03:05:39 +00:00
parent 6ebf33fa39
commit 909f71d9e0
8 changed files with 259 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fastlib/u/garryb/nbr/build.py
+18 -11
View File
@@ -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"],
fastlib/u/garryb/nbr/nbr_utils.h
+4 -1
View File
@@ -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<typename GNP, typename Solver>
class RpcMonochromaticDualTreeRunner {
@@ -504,7 +505,9 @@ void RpcMonochromaticDualTreeMain(datanode *module, const char *gnp_name) {
RpcMonochromaticDualTreeRunner<GNP, Solver> runner;
runner.Main(module, gnp_name);
}
#endif
};
#endif
fastlib/u/garryb/nbr/netcache.h
+3 -3
View File
@@ -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<T> {
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<T> {
small_cache_.Clear(mode);
}
RawRemoteObjectBackend *server() {
Channel *server() {
return remote_device_.server();
}
fastlib/u/garryb/nbr/rpc.cc
+23 -16
View File
@@ -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) {
fastlib/u/garryb/nbr/rpc.h
+9 -8
View File
@@ -24,6 +24,7 @@ template<class ResponseObject>
class Rpc {
FORBID_COPY(Rpc);
private:
template<class RequestObject>
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<typename RequestObject>
ResponseObject *Request(
int channel, int peer, const RequestObject& request) {
RpcRequestTransaction transaction;
RpcRequestTransaction<RequestObject> transaction;
response_ = transaction.Doit(channel, peer, request);
response_object_ = ot::PointerThaw<ResponseObject>(response_.data());
response_object_ = ot::PointerThaw<ResponseObject>(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<typename RequestObject, typename ResponseObject>
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);
}
fastlib/u/garryb/nbr/rpc_sock.cc
+183 -133
View File
@@ -4,6 +4,15 @@
* Implementation of transaction API using TCP.
*/
#include "rpc.h"
#include "rpc_sock.h"
#include "fastlib/fastlib.h"
#include <fcntl.h>
#include <errno.h>
#include <string.h>
/*
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<WorkItem*> *work_items) {
ArrayList<WorkItem> *work_items) {
ArrayList<Message*>* 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<char*>(&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<char>(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<char>(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);
}
}
fastlib/u/garryb/nbr/rpc_sock.h
+17 -10
View File
@@ -7,6 +7,8 @@
#ifndef RPC_SOCK_H
#define RPC_SOCK_H
#include "fastlib/fastlib_int.h"
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/select.h>
@@ -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<PeerInfo> 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<Message*> read_queue_;
int write_total;
int write_total_;
Message *write_message_;
size_t write_buffer_pos_;
MinHeap<int, Message*> 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<WorkItem*>* work_items);
void GatherReadyMessages_(Peer *peer, ArrayList<WorkItem>* work_items);
};
namespace rpc {
fastlib/u/garryb/nbr/rpc_sock_test.cc
+2
View File
@@ -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();
}