From fd0af5b6efb1e3b61300047e8a54e0a4e6ef10e2 Mon Sep 17 00:00:00 2001 From: David Rose Date: Sat, 13 Sep 2008 03:39:21 +0000 Subject: [PATCH] fix pstats-threaded-write in simple-threads case --- panda/src/nativenet/socket_fdset.h | 2 +- panda/src/nativenet/socket_tcp.h | 5 ++- panda/src/net/connection.cxx | 15 ++++--- panda/src/net/connectionManager.cxx | 33 +++++++++++++- panda/src/net/connectionReader.cxx | 5 +-- panda/src/net/connectionWriter.cxx | 53 +++++++++++++++++++---- panda/src/net/connectionWriter.h | 12 +++-- panda/src/net/datagramQueue.cxx | 18 +++++++- panda/src/net/datagramQueue.h | 4 +- panda/src/pstatclient/config_pstats.cxx | 9 ++++ panda/src/pstatclient/config_pstats.h | 1 + panda/src/pstatclient/pStatClientImpl.cxx | 13 +++--- 12 files changed, 135 insertions(+), 35 deletions(-) diff --git a/panda/src/nativenet/socket_fdset.h b/panda/src/nativenet/socket_fdset.h index 40d1d72069..2c848ae582 100755 --- a/panda/src/nativenet/socket_fdset.h +++ b/panda/src/nativenet/socket_fdset.h @@ -18,7 +18,7 @@ class Socket_fdset { -public: +PUBLISHED: inline Socket_fdset(); inline void setForSocket(const Socket_IP &incon); diff --git a/panda/src/nativenet/socket_tcp.h b/panda/src/nativenet/socket_tcp.h index a0e535e927..b3c4555736 100755 --- a/panda/src/nativenet/socket_tcp.h +++ b/panda/src/nativenet/socket_tcp.h @@ -158,14 +158,15 @@ bool Socket_TCP::ActiveOpenNonBlocking(const Socket_Address & theaddress) SetNonBlocking(); SetReuseAddress(); - if (DO_CONNECT(_socket, &theaddress.GetAddressInfo()) != 0) + if (DO_CONNECT(_socket, &theaddress.GetAddressInfo()) != 0) { if (GETERROR() != LOCAL_CONNECT_BLOCKING) { printf(" None Blockign Connect Error %d",GETERROR()); return ErrorClose(); } + } - return true; + return true; } //////////////////////////////////////////////////////////////////// diff --git a/panda/src/net/connection.cxx b/panda/src/net/connection.cxx index e701c48b6e..43bfc5a13b 100644 --- a/panda/src/net/connection.cxx +++ b/panda/src/net/connection.cxx @@ -367,8 +367,13 @@ send_datagram(const NetDatagram &datagram, int tcp_header_size) { data += header.get_header(); data += datagram.get_message(); + if (net_cat.is_debug()) { + header.verify_datagram(datagram); + } + int bytes_to_send = data.length(); Socket_Address addr = datagram.get_address().get_addr(); + bool okflag = udp->SendTo(data, addr); #if defined(HAVE_THREADS) && defined(SIMPLE_THREADS) while (!okflag && udp->GetLastError() == LOCAL_BLOCKING_ERROR) { @@ -376,14 +381,10 @@ send_datagram(const NetDatagram &datagram, int tcp_header_size) { okflag = udp->SendTo(data, addr); } #endif // SIMPLE_THREADS - - if (net_cat.is_debug()) { - header.verify_datagram(datagram); - } - + if (net_cat.is_spam()) { net_cat.spam() - << "Sending UDP datagram with " + << "Sent UDP datagram with " << bytes_to_send << " bytes to " << (void *)this << ", ok = " << okflag << "\n"; } @@ -449,7 +450,7 @@ send_raw_datagram(const NetDatagram &datagram) { if (net_cat.is_spam()) { net_cat.spam() - << "Sending UDP datagram with " + << "Sent UDP datagram with " << data.size() << " bytes to " << (void *)this << ", ok = " << okflag << "\n"; } diff --git a/panda/src/net/connectionManager.cxx b/panda/src/net/connectionManager.cxx index ed897c9fe8..8df1e5f4b3 100644 --- a/panda/src/net/connectionManager.cxx +++ b/panda/src/net/connectionManager.cxx @@ -19,6 +19,7 @@ #include "netAddress.h" #include "config_net.h" #include "mutexHolder.h" +#include "trueClock.h" #ifdef WIN32_VC #include // For gethostname() @@ -145,7 +146,37 @@ open_TCP_server_rendezvous(int port, int backlog) { PT(Connection) ConnectionManager:: open_TCP_client_connection(const NetAddress &address, int timeout_ms) { Socket_TCP *socket = new Socket_TCP; - bool okflag = socket->ActiveOpen(address.get_addr(),true); + +#if defined(HAVE_THREADS) && defined(SIMPLE_THREADS) + // In the simple-threads case, use a non-blocking connect. + bool okflag = socket->ActiveOpenNonBlocking(address.get_addr()); + if (okflag && socket->GetLastError() == LOCAL_CONNECT_BLOCKING) { + // Now wait for the socket to connect. + TrueClock *clock = TrueClock::get_global_ptr(); + double start = clock->get_short_time(); + Thread::force_yield(); + Socket_fdset fset; + fset.setForSocket(*socket); + int ready = fset.WaitForWrite(true, 0); + while (ready == 0) { + double elapsed = clock->get_short_time() - start; + if (elapsed * 1000.0 > timeout_ms) { + // Timeout. + okflag = false; + break; + } + Thread::force_yield(); + fset.setForSocket(*socket); + ready = fset.WaitForWrite(true, 0); + } + } + +#else + // In the normal case, we can just do a blocking connect. + bool okflag = socket->ActiveOpen(address.get_addr(), false); + +#endif // SIMPLE_THREADS + if (!okflag) { net_cat.error() << "Unable to open TCP connection to server " diff --git a/panda/src/net/connectionReader.cxx b/panda/src/net/connectionReader.cxx index a57056f647..cf074e6c01 100644 --- a/panda/src/net/connectionReader.cxx +++ b/panda/src/net/connectionReader.cxx @@ -98,10 +98,7 @@ ConnectionReader:: ConnectionReader(ConnectionManager *manager, int num_threads) : _manager(manager) { - if (!Thread::is_true_threads()) { - // There is no point in using threads for this kind of I/O unless - // we actually have real threads available (i.e. HAVE_THREADS is - // defined, and SIMPLE_THREADS is not). + if (!Thread::is_threading_supported()) { #ifndef NDEBUG if (num_threads != 0) { if (net_cat.is_debug()) { diff --git a/panda/src/net/connectionWriter.cxx b/panda/src/net/connectionWriter.cxx index 331b3a4493..ac6dcf20eb 100644 --- a/panda/src/net/connectionWriter.cxx +++ b/panda/src/net/connectionWriter.cxx @@ -57,10 +57,7 @@ ConnectionWriter:: ConnectionWriter(ConnectionManager *manager, int num_threads) : _manager(manager) { - if (!Thread::is_true_threads()) { - // There is no point in using threads for this kind of I/O unless - // we actually have real threads available (i.e. HAVE_THREADS is - // defined, and SIMPLE_THREADS is not). + if (!Thread::is_threading_supported()) { #ifndef NDEBUG if (num_threads != 0) { if (net_cat.is_debug()) { @@ -109,6 +106,38 @@ ConnectionWriter:: (*ti)->join(); } } +// Function: ConnectionWriter::set_max_queue_size +// Access: Public +// Description: Limits the number of packets that may be pending on +// the outbound queue. This only has an effect when +// using threads; if num_threads is 0, then all packets +// are sent immediately. +//////////////////////////////////////////////////////////////////// +void ConnectionWriter:: +set_max_queue_size(int max_size) { + _queue.set_max_queue_size(max_size); +} + +//////////////////////////////////////////////////////////////////// +// Function: ConnectionWriter::get_max_queue_size +// Access: Public +// Description: Returns the maximum size the queue is allowed to grow +// to. See set_max_queue_size(). +//////////////////////////////////////////////////////////////////// +int ConnectionWriter:: +get_max_queue_size() const { + return _queue.get_max_queue_size(); +} + +//////////////////////////////////////////////////////////////////// +// Function: ConnectionWriter::get_current_queue_size +// Access: Public +// Description: Returns the current number of things in the queue. +//////////////////////////////////////////////////////////////////// +int ConnectionWriter:: +get_current_queue_size() const { + return _queue.get_current_queue_size(); +} //////////////////////////////////////////////////////////////////// @@ -125,9 +154,13 @@ ConnectionWriter:: // only returns false if the send queue is filled; it's // impossible to detect a transmission error at this // point. +// +// If block is true, this will not return false if the +// send queue is filled; instead, it will wait until +// this is space available. //////////////////////////////////////////////////////////////////// bool ConnectionWriter:: -send(const Datagram &datagram, const PT(Connection) &connection) { +send(const Datagram &datagram, const PT(Connection) &connection, bool block) { nassertr(connection != (Connection *)NULL, false); nassertr(connection->get_socket()->is_exact_type(Socket_TCP::get_class_type()), false); @@ -141,7 +174,7 @@ send(const Datagram &datagram, const PT(Connection) &connection) { return connection->send_datagram(copy, _tcp_header_size); } } else { - return _queue.insert(copy); + return _queue.insert(copy, block); } } @@ -160,10 +193,14 @@ send(const Datagram &datagram, const PT(Connection) &connection) { // only returns false if the send queue is filled; it's // impossible to detect a transmission error at this // point. +// +// If block is true, this will not return false if the +// send queue is filled; instead, it will wait until +// this is space available. //////////////////////////////////////////////////////////////////// bool ConnectionWriter:: send(const Datagram &datagram, const PT(Connection) &connection, - const NetAddress &address) { + const NetAddress &address, bool block) { nassertr(connection != (Connection *)NULL, false); nassertr(connection->get_socket()->is_exact_type(Socket_UDP::get_class_type()), false); @@ -186,7 +223,7 @@ send(const Datagram &datagram, const PT(Connection) &connection, return connection->send_datagram(copy, _tcp_header_size); } } else { - return _queue.insert(copy); + return _queue.insert(copy, block); } } diff --git a/panda/src/net/connectionWriter.h b/panda/src/net/connectionWriter.h index 4acf169e4a..9ab4c875e1 100644 --- a/panda/src/net/connectionWriter.h +++ b/panda/src/net/connectionWriter.h @@ -40,12 +40,18 @@ PUBLISHED: ConnectionWriter(ConnectionManager *manager, int num_threads); ~ConnectionWriter(); - bool send(const Datagram &datagram, - const PT(Connection) &connection); + void set_max_queue_size(int max_size); + int get_max_queue_size() const; + int get_current_queue_size() const; bool send(const Datagram &datagram, const PT(Connection) &connection, - const NetAddress &address); + bool block = false); + + bool send(const Datagram &datagram, + const PT(Connection) &connection, + const NetAddress &address, + bool block = false); bool is_valid_for_udp(const Datagram &datagram) const; diff --git a/panda/src/net/datagramQueue.cxx b/panda/src/net/datagramQueue.cxx index c5ce25044c..45fb1b80a8 100644 --- a/panda/src/net/datagramQueue.cxx +++ b/panda/src/net/datagramQueue.cxx @@ -68,16 +68,27 @@ shutdown() { // threads are waiting on the queue, this will wake one // of them up. Returns true if successful, false if the // queue was full. +// +// If block is true, this will not return until +// successful, waiting until the queue has space +// available if necessary. //////////////////////////////////////////////////////////////////// bool DatagramQueue:: -insert(const NetDatagram &data) { +insert(const NetDatagram &data, bool block) { MutexHolder holder(_cvlock); bool enqueue_ok = ((int)_queue.size() < _max_queue_size); + if (block) { + while (!enqueue_ok && !_shutdown) { + _cv.wait(); + enqueue_ok = ((int)_queue.size() < _max_queue_size); + } + } + if (enqueue_ok) { _queue.push_back(data); } - _cv.signal(); + _cv.signal(); // Only need to wake up one thread. return enqueue_ok; } @@ -121,6 +132,9 @@ extract(NetDatagram &result) { result = _queue.front(); _queue.pop_front(); + // Wake up any threads waiting to stuff things into the queue. + _cv.signal_all(); + return true; } diff --git a/panda/src/net/datagramQueue.h b/panda/src/net/datagramQueue.h index de0dc28d76..3db8f754e4 100644 --- a/panda/src/net/datagramQueue.h +++ b/panda/src/net/datagramQueue.h @@ -34,7 +34,7 @@ public: ~DatagramQueue(); void shutdown(); - bool insert(const NetDatagram &data); + bool insert(const NetDatagram &data, bool block = false); bool extract(NetDatagram &result); void set_max_queue_size(int max_size); @@ -43,7 +43,7 @@ public: private: Mutex _cvlock; - ConditionVarFull _cv; + ConditionVarFull _cv; // signaled when queue contents change. typedef pdeque QueueType; QueueType _queue; diff --git a/panda/src/pstatclient/config_pstats.cxx b/panda/src/pstatclient/config_pstats.cxx index 8780401aae..7a78718c86 100644 --- a/panda/src/pstatclient/config_pstats.cxx +++ b/panda/src/pstatclient/config_pstats.cxx @@ -39,6 +39,15 @@ ConfigVariableBool pstats_threaded_write "wouldn't want this set true, unless you suspect something is " "broken with the threaded network interfaces.")); +ConfigVariableInt pstats_max_queue_size +("pstats-max-queue-size", 1, + PRC_DESC("If pstats-threaded-write is true, this specifies the maximum " + "number of packets (generally, frames of data) that may be queued " + "up for the thread to process. If this is large, the writer " + "thread may fall behind and the output of PStats will lag. Keep " + "this small to drop missed packets on the floor instead, and " + "ensure that the frame data does not grow stale.")); + ConfigVariableDouble pstats_tcp_ratio ("pstats-tcp-ratio", 0.01, PRC_DESC("This specifies the ratio of frame update messages that are eligible " diff --git a/panda/src/pstatclient/config_pstats.h b/panda/src/pstatclient/config_pstats.h index 04bde59652..2d818af324 100644 --- a/panda/src/pstatclient/config_pstats.h +++ b/panda/src/pstatclient/config_pstats.h @@ -32,6 +32,7 @@ NotifyCategoryDecl(pstats, EXPCL_PANDA_PSTATCLIENT, EXPTP_PANDA_PSTATCLIENT); extern EXPCL_PANDA_PSTATCLIENT ConfigVariableString pstats_name; extern EXPCL_PANDA_PSTATCLIENT ConfigVariableDouble pstats_max_rate; extern EXPCL_PANDA_PSTATCLIENT ConfigVariableBool pstats_threaded_write; +extern EXPCL_PANDA_PSTATCLIENT ConfigVariableInt pstats_max_queue_size; extern EXPCL_PANDA_PSTATCLIENT ConfigVariableDouble pstats_tcp_ratio; extern EXPCL_PANDA_PSTATCLIENT ConfigVariableString pstats_host; diff --git a/panda/src/pstatclient/pStatClientImpl.cxx b/panda/src/pstatclient/pStatClientImpl.cxx index 27cb1f5e62..93988b7aa7 100644 --- a/panda/src/pstatclient/pStatClientImpl.cxx +++ b/panda/src/pstatclient/pStatClientImpl.cxx @@ -47,6 +47,7 @@ PStatClientImpl(PStatClient *client) : _reader(this, 0), _writer(this, pstats_threaded_write ? 1 : 0) { + _writer.set_max_queue_size(pstats_max_queue_size); _reader.set_tcp_header_size(4); _writer.set_tcp_header_size(4); _is_connected = false; @@ -289,8 +290,10 @@ transmit_frame_data(int thread_index) { thread->_next_packet = now + packet_delay; if (!sent) { - pstats_cat.debug() - << "Couldn't send packet.\n"; + if (pstats_cat.is_debug()) { + pstats_cat.debug() + << "Couldn't send packet.\n"; + } } } } @@ -363,7 +366,7 @@ send_hello() { Datagram datagram; message.encode(datagram); - _writer.send(datagram, _tcp_connection); + _writer.send(datagram, _tcp_connection, true); } //////////////////////////////////////////////////////////////////// @@ -393,7 +396,7 @@ report_new_collectors() { Datagram datagram; message.encode(datagram); - _writer.send(datagram, _tcp_connection); + _writer.send(datagram, _tcp_connection, true); } } @@ -418,7 +421,7 @@ report_new_threads() { Datagram datagram; message.encode(datagram); - _writer.send(datagram, _tcp_connection); + _writer.send(datagram, _tcp_connection, true); } }