396 lines
14 KiB
C++
396 lines
14 KiB
C++
// Filename: connectionWriter.cxx
|
|
// Created by: drose (08Feb00)
|
|
//
|
|
////////////////////////////////////////////////////////////////////
|
|
//
|
|
// PANDA 3D SOFTWARE
|
|
// Copyright (c) Carnegie Mellon University. All rights reserved.
|
|
//
|
|
// All use of this software is subject to the terms of the revised BSD
|
|
// license. You should have received a copy of this license along
|
|
// with this source code in a file named "LICENSE."
|
|
//
|
|
////////////////////////////////////////////////////////////////////
|
|
|
|
#include "connectionWriter.h"
|
|
#include "connectionManager.h"
|
|
#include "datagramTCPHeader.h"
|
|
#include "config_net.h"
|
|
#include "socket_tcp.h"
|
|
#include "socket_udp.h"
|
|
#include "pnotify.h"
|
|
#include "config_downloader.h"
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::WriterThread::Constructor
|
|
// Access: Public
|
|
// Description:
|
|
////////////////////////////////////////////////////////////////////
|
|
ConnectionWriter::WriterThread::
|
|
WriterThread(ConnectionWriter *writer, const string &thread_name,
|
|
int thread_index) :
|
|
Thread(make_thread_name(thread_name, thread_index),
|
|
make_thread_name(thread_name, thread_index)),
|
|
_writer(writer),
|
|
_thread_index(thread_index)
|
|
{
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::WriterThread::thread_main
|
|
// Access: Public, Virtual
|
|
// Description:
|
|
////////////////////////////////////////////////////////////////////
|
|
void ConnectionWriter::WriterThread::
|
|
thread_main() {
|
|
_writer->thread_run(_thread_index);
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::Constructor
|
|
// Access: Public
|
|
// Description: Creates a new ConnectionWriter with the indicated
|
|
// number of threads to handle output.
|
|
//
|
|
// If num_threads is 0, all datagrams will be sent
|
|
// immediately instead of queueing for later
|
|
// transmission by a thread.
|
|
////////////////////////////////////////////////////////////////////
|
|
ConnectionWriter::
|
|
ConnectionWriter(ConnectionManager *manager, int num_threads,
|
|
const string &thread_name) :
|
|
_manager(manager)
|
|
{
|
|
if (!Thread::is_threading_supported()) {
|
|
#ifndef NDEBUG
|
|
if (num_threads != 0) {
|
|
if (net_cat.is_debug()) {
|
|
net_cat.debug()
|
|
<< "Threading support is not available.\n";
|
|
}
|
|
}
|
|
#endif // NDEBUG
|
|
num_threads = 0;
|
|
}
|
|
|
|
_raw_mode = false;
|
|
_tcp_header_size = tcp_header_size;
|
|
_immediate = (num_threads <= 0);
|
|
_shutdown = false;
|
|
|
|
string writer_thread_name = thread_name;
|
|
if (thread_name.empty()) {
|
|
writer_thread_name = "WriterThread";
|
|
}
|
|
int i;
|
|
for (i = 0; i < num_threads; i++) {
|
|
PT(WriterThread) thread = new WriterThread(this, writer_thread_name, i);
|
|
_threads.push_back(thread);
|
|
}
|
|
for (i = 0; i < num_threads; i++) {
|
|
_threads[i]->start(net_thread_priority, true);
|
|
}
|
|
|
|
_manager->add_writer(this);
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::Destructor
|
|
// Access: Public
|
|
// Description:
|
|
////////////////////////////////////////////////////////////////////
|
|
ConnectionWriter::
|
|
~ConnectionWriter() {
|
|
if (_manager != (ConnectionManager *)NULL) {
|
|
_manager->remove_writer(this);
|
|
}
|
|
|
|
shutdown();
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// 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();
|
|
}
|
|
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::send
|
|
// Access: Public
|
|
// Description: Enqueues a datagram for transmittal on the indicated
|
|
// socket. Since the host address is not specified with
|
|
// this form, this function should only be used for
|
|
// sending TCP packets. Use the other send() method for
|
|
// sending UDP packets.
|
|
//
|
|
// Returns true if successful, false if there was an
|
|
// error. In the normal, threaded case, this function
|
|
// 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
|
|
// there is space available.
|
|
////////////////////////////////////////////////////////////////////
|
|
bool ConnectionWriter::
|
|
send(const Datagram &datagram, const PT(Connection) &connection, bool block) {
|
|
nassertr(!_shutdown, false);
|
|
nassertr(connection != (Connection *)NULL, false);
|
|
nassertr(connection->get_socket()->is_exact_type(Socket_TCP::get_class_type()), false);
|
|
|
|
NetDatagram copy(datagram);
|
|
copy.set_connection(connection);
|
|
|
|
if (_immediate) {
|
|
if (_raw_mode) {
|
|
return connection->send_raw_datagram(copy);
|
|
} else {
|
|
return connection->send_datagram(copy, _tcp_header_size);
|
|
}
|
|
} else {
|
|
return _queue.insert(copy, block);
|
|
}
|
|
}
|
|
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::send
|
|
// Access: Public
|
|
// Description: Enqueues a datagram for transmittal on the indicated
|
|
// socket. This form of the function allows the
|
|
// specification of a destination host address, and so
|
|
// is appropriate for UDP packets. Use the other send()
|
|
// method for sending TCP packets.
|
|
//
|
|
// Returns true if successful, false if there was an
|
|
// error. In the normal, threaded case, this function
|
|
// 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
|
|
// there is space available.
|
|
////////////////////////////////////////////////////////////////////
|
|
bool ConnectionWriter::
|
|
send(const Datagram &datagram, const PT(Connection) &connection,
|
|
const NetAddress &address, bool block) {
|
|
nassertr(!_shutdown, false);
|
|
nassertr(connection != (Connection *)NULL, false);
|
|
nassertr(connection->get_socket()->is_exact_type(Socket_UDP::get_class_type()), false);
|
|
|
|
if ((int)datagram.get_length() > maximum_udp_datagram) {
|
|
net_cat.warning()
|
|
<< "Attempt to send UDP datagram of " << datagram.get_length()
|
|
<< " bytes, more than the\n"
|
|
<< "currently defined maximum of " << maximum_udp_datagram
|
|
<< " bytes.\n";
|
|
}
|
|
|
|
NetDatagram copy(datagram);
|
|
copy.set_connection(connection);
|
|
copy.set_address(address);
|
|
|
|
if (_immediate) {
|
|
if (_raw_mode) {
|
|
return connection->send_raw_datagram(copy);
|
|
} else {
|
|
return connection->send_datagram(copy, _tcp_header_size);
|
|
}
|
|
} else {
|
|
return _queue.insert(copy, block);
|
|
}
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::is_valid_for_udp
|
|
// Access: Public
|
|
// Description: Returns true if the datagram is small enough to be
|
|
// sent over a UDP packet, false otherwise.
|
|
////////////////////////////////////////////////////////////////////
|
|
bool ConnectionWriter::
|
|
is_valid_for_udp(const Datagram &datagram) const {
|
|
return (int)datagram.get_length() <= maximum_udp_datagram;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::get_manager
|
|
// Access: Public
|
|
// Description: Returns a pointer to the ConnectionManager object
|
|
// that serves this ConnectionWriter.
|
|
////////////////////////////////////////////////////////////////////
|
|
ConnectionManager *ConnectionWriter::
|
|
get_manager() const {
|
|
return _manager;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::is_immediate
|
|
// Access: Public
|
|
// Description: Returns true if the writer is an immediate writer,
|
|
// i.e. it has no threads.
|
|
////////////////////////////////////////////////////////////////////
|
|
bool ConnectionWriter::
|
|
is_immediate() const {
|
|
return _immediate;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::get_num_threads
|
|
// Access: Public
|
|
// Description: Returns the number of threads the ConnectionWriter
|
|
// has been created with.
|
|
////////////////////////////////////////////////////////////////////
|
|
int ConnectionWriter::
|
|
get_num_threads() const {
|
|
return _threads.size();
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::set_raw_mode
|
|
// Access: Public
|
|
// Description: Sets the ConnectionWriter into raw mode (or turns off
|
|
// raw mode). In raw mode, datagrams are not sent along
|
|
// with their headers; the bytes in the datagram are
|
|
// simply sent down the pipe.
|
|
//
|
|
// Setting the ConnectionWriter to raw mode must be done
|
|
// with care. This can only be done when the matching
|
|
// ConnectionReader is also set to raw mode, or when the
|
|
// ConnectionWriter is communicating to a process that
|
|
// does not expect datagrams.
|
|
////////////////////////////////////////////////////////////////////
|
|
void ConnectionWriter::
|
|
set_raw_mode(bool mode) {
|
|
_raw_mode = mode;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::get_raw_mode
|
|
// Access: Public
|
|
// Description: Returns the current setting of the raw mode flag.
|
|
// See set_raw_mode().
|
|
////////////////////////////////////////////////////////////////////
|
|
bool ConnectionWriter::
|
|
get_raw_mode() const {
|
|
return _raw_mode;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::set_tcp_header_size
|
|
// Access: Public
|
|
// Description: Sets the header size of TCP packets. At the present,
|
|
// legal values for this are 0, 2, or 4; this specifies
|
|
// the number of bytes to use encode the datagram length
|
|
// at the start of each TCP datagram. Sender and
|
|
// receiver must independently agree on this.
|
|
////////////////////////////////////////////////////////////////////
|
|
void ConnectionWriter::
|
|
set_tcp_header_size(int tcp_header_size) {
|
|
_tcp_header_size = tcp_header_size;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::get_tcp_header_size
|
|
// Access: Public
|
|
// Description: Returns the current setting of TCP header size.
|
|
// See set_tcp_header_size().
|
|
////////////////////////////////////////////////////////////////////
|
|
int ConnectionWriter::
|
|
get_tcp_header_size() const {
|
|
return _tcp_header_size;
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::shutdown
|
|
// Access: Published
|
|
// Description: Stops all the threads and cleans them up. This is
|
|
// called automatically by the destructor, but it may be
|
|
// called explicitly before destruction.
|
|
////////////////////////////////////////////////////////////////////
|
|
void ConnectionWriter::
|
|
shutdown() {
|
|
if (_shutdown) {
|
|
return;
|
|
}
|
|
_shutdown = true;
|
|
|
|
// First, shutdown the queue. This will tell our threads they're
|
|
// done.
|
|
_queue.shutdown();
|
|
|
|
// Now wait for all threads to terminate.
|
|
Threads::iterator ti;
|
|
for (ti = _threads.begin(); ti != _threads.end(); ++ti) {
|
|
(*ti)->join();
|
|
}
|
|
_threads.clear();
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::clear_manager
|
|
// Access: Protected
|
|
// Description: This should normally only be called when the
|
|
// associated ConnectionManager destructs. It resets
|
|
// the ConnectionManager pointer to NULL so we don't
|
|
// have a floating pointer. This makes the
|
|
// ConnectionWriter invalid; presumably it also will be
|
|
// destructed momentarily.
|
|
////////////////////////////////////////////////////////////////////
|
|
void ConnectionWriter::
|
|
clear_manager() {
|
|
_manager = (ConnectionManager *)NULL;
|
|
shutdown();
|
|
}
|
|
|
|
////////////////////////////////////////////////////////////////////
|
|
// Function: ConnectionWriter::thread_run
|
|
// Access: Private
|
|
// Description: This is the actual executing function for each
|
|
// thread.
|
|
////////////////////////////////////////////////////////////////////
|
|
void ConnectionWriter::
|
|
thread_run(int thread_index) {
|
|
nassertv(!_immediate);
|
|
|
|
NetDatagram datagram;
|
|
while (_queue.extract(datagram)) {
|
|
if (_raw_mode) {
|
|
datagram.get_connection()->send_raw_datagram(datagram);
|
|
} else {
|
|
datagram.get_connection()->send_datagram(datagram, _tcp_header_size);
|
|
}
|
|
Thread::consider_yield();
|
|
}
|
|
}
|