aboutsummaryrefslogtreecommitdiff
path: root/nxcomp/Proxy.cpp
diff options
context:
space:
mode:
authorReinhard Tartler <siretart@tauware.de>2011-11-13 09:27:51 +0100
committerReinhard Tartler <siretart@tauware.de>2011-11-13 09:27:51 +0100
commit3e924126b56c4d421e8263d25f6b14aa4ceed047 (patch)
tree921403de7b834fb1ac89f97d84ce105e0b487051 /nxcomp/Proxy.cpp
parenta840692edc9c6d19cd7c057f68e39c7d95eb767d (diff)
downloadnx-libs-3e924126b56c4d421e8263d25f6b14aa4ceed047.tar.gz
nx-libs-3e924126b56c4d421e8263d25f6b14aa4ceed047.tar.bz2
nx-libs-3e924126b56c4d421e8263d25f6b14aa4ceed047.zip
Imported nxcomp-3.1.0-4.tar.gznxcomp/3.1.0-4
Summary: Imported nxcomp-3.1.0-4.tar.gz Keywords: Imported nxcomp-3.1.0-4.tar.gz into Git repository
Diffstat (limited to 'nxcomp/Proxy.cpp')
-rwxr-xr-xnxcomp/Proxy.cpp6444
1 files changed, 6444 insertions, 0 deletions
diff --git a/nxcomp/Proxy.cpp b/nxcomp/Proxy.cpp
new file mode 100755
index 000000000..371dfe645
--- /dev/null
+++ b/nxcomp/Proxy.cpp
@@ -0,0 +1,6444 @@
+/**************************************************************************/
+/* */
+/* Copyright (c) 2001, 2007 NoMachine, http://www.nomachine.com/. */
+/* */
+/* NXCOMP, NX protocol compression and NX extensions to this software */
+/* are copyright of NoMachine. Redistribution and use of the present */
+/* software is allowed according to terms specified in the file LICENSE */
+/* which comes in the source distribution. */
+/* */
+/* Check http://www.nomachine.com/licensing.html for applicability. */
+/* */
+/* NX and NoMachine are trademarks of NoMachine S.r.l. */
+/* */
+/* All rights reserved. */
+/* */
+/**************************************************************************/
+
+#include <stdio.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <fstream.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+
+#if defined(__CYGWIN32__) || defined(__APPLE__) || defined(__FreeBSD__) || defined(__sun)
+#include <netinet/in_systm.h>
+#endif
+
+#ifndef __CYGWIN32__
+#include <sys/un.h>
+#endif
+
+#include <netinet/in.h>
+#include <netinet/ip.h>
+#include <netinet/tcp.h>
+
+#if defined(__EMX__ ) || defined(__CYGWIN32__)
+
+struct sockaddr_un
+{
+ u_short sun_family;
+ char sun_path[108];
+};
+
+#endif
+
+#include "NXalert.h"
+#include "NXvars.h"
+
+#include "Proxy.h"
+
+#include "Socket.h"
+#include "Channel.h"
+#include "Statistics.h"
+
+#include "ClientChannel.h"
+#include "ServerChannel.h"
+#include "GenericChannel.h"
+
+//
+// We need to adjust some values related
+// to these messages at the time the mes-
+// sage stores are reconfigured.
+//
+
+#include "PutImage.h"
+#include "ChangeGC.h"
+#include "PolyFillRectangle.h"
+#include "PutPackedImage.h"
+
+//
+// This is from the main loop.
+//
+
+extern void CleanupListeners();
+
+//
+// Default size of string buffers.
+//
+
+#define DEFAULT_STRING_LENGTH 512
+
+//
+// Set the verbosity level. You also need
+// to define DUMP in Misc.cpp if DUMP is
+// defined here.
+//
+
+#define WARNING
+#define PANIC
+#undef TEST
+#undef DEBUG
+#undef DUMP
+
+//
+// Log the important tracepoints related
+// to writing packets to the peer proxy.
+//
+
+#undef FLUSH
+
+//
+// Log the operations related to splits.
+//
+
+#undef SPLIT
+
+//
+// Log the operations related to sending
+// and receiving the control tokens.
+//
+
+#undef TOKEN
+
+//
+// Log the operations related to setting
+// the token limits.
+//
+
+#undef LIMIT
+
+//
+// Log a warning if no data is written by
+// the proxy within a timeout.
+//
+
+#undef TIME
+
+//
+// Log the operation related to generating
+// the ping message at idle time.
+//
+
+#undef PING
+
+Proxy::Proxy(int fd)
+
+ : transport_(new ProxyTransport(fd)), fd_(fd), readBuffer_(transport_)
+{
+ for (int channelId = 0;
+ channelId < CONNECTIONS_LIMIT;
+ channelId++)
+ {
+ channels_[channelId] = NULL;
+ transports_[channelId] = NULL;
+ congestions_[channelId] = 0;
+
+ fdMap_[channelId] = nothing;
+ channelMap_[channelId] = nothing;
+ }
+
+ inputChannel_ = nothing;
+ outputChannel_ = nothing;
+
+ controlLength_ = 0;
+
+ operation_ = operation_in_negotiation;
+
+ draining_ = 0;
+ priority_ = 0;
+ finish_ = 0;
+ shutdown_ = 0;
+ congestion_ = 0;
+
+ timer_ = 0;
+ alert_ = 0;
+
+ agent_ = nothing;
+
+ //
+ // Set null timeouts. This will require
+ // a new link configuration.
+ //
+
+ timeouts_.split = 0;
+ timeouts_.motion = 0;
+
+ timeouts_.readTs = getTimestamp();
+ timeouts_.writeTs = getTimestamp();
+
+ timeouts_.loopTs = getTimestamp();
+ timeouts_.pingTs = getTimestamp();
+ timeouts_.alertTs = nullTimestamp();
+ timeouts_.loadTs = nullTimestamp();
+
+ timeouts_.splitTs = nullTimestamp();
+ timeouts_.motionTs = nullTimestamp();
+
+ //
+ // Initialize the token counters. This
+ // will require a new link configuration.
+ //
+
+ for (int i = token_control; i <= token_data; i++)
+ {
+ tokens_[i].size = 0;
+ tokens_[i].limit = 0;
+
+ tokens_[i].bytes = 0;
+ tokens_[i].remaining = 0;
+ }
+
+ tokens_[token_control].request = code_control_token_request;
+ tokens_[token_control].reply = code_control_token_reply;
+ tokens_[token_control].type = token_control;
+
+ tokens_[token_split].request = code_split_token_request;
+ tokens_[token_split].reply = code_split_token_reply;
+ tokens_[token_split].type = token_split;
+
+ tokens_[token_data].request = code_data_token_request;
+ tokens_[token_data].reply = code_data_token_reply;
+ tokens_[token_data].type = token_data;
+
+ currentStatistics_ = NULL;
+
+ //
+ // Create compressor and decompressor
+ // for image and data payload.
+ //
+
+ compressor_ = new StaticCompressor(control -> LocalDataCompressionLevel,
+ control -> LocalDataCompressionThreshold);
+
+ //
+ // Create object storing NX specific
+ // opcodes.
+ //
+
+ opcodeStore_ = new OpcodeStore();
+
+ //
+ // Create the message stores.
+ //
+
+ clientStore_ = new ClientStore(compressor_);
+ serverStore_ = new ServerStore(compressor_);
+
+ //
+ // Older proxies will refuse to store
+ // messages bigger than 262144 bytes.
+ //
+
+ if (control -> isProtoStep7() == 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: WARNING! Limiting the maximum "
+ << "message size to " << 262144 << ".\n"
+ << logofs_flush;
+ #endif
+
+ control -> MaximumMessageSize = 262144;
+ }
+
+ clientCache_ = new ClientCache();
+ serverCache_ = new ServerCache();
+
+ if (clientCache_ == NULL || serverCache_ == NULL)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Failed to create the channel cache.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Failed to create the channel cache.\n";
+
+ HandleCleanup();
+ }
+
+ //
+ // Prepare for image decompression.
+ //
+
+ UnpackInit();
+
+ #ifdef DEBUG
+ *logofs << "Proxy: Created new object at " << this
+ << ".\n" << logofs_flush;
+ #endif
+}
+
+Proxy::~Proxy()
+{
+ for (int channelId = 0;
+ channelId < CONNECTIONS_LIMIT;
+ channelId++)
+ {
+ if (channels_[channelId] != NULL)
+ {
+ deallocateTransport(channelId);
+
+ delete channels_[channelId];
+ channels_[channelId] = NULL;
+ }
+ }
+
+ delete transport_;
+ delete compressor_;
+
+ //
+ // Delete storage shared among channels.
+ //
+
+ delete opcodeStore_;
+
+ delete clientStore_;
+ delete serverStore_;
+
+ delete clientCache_;
+ delete serverCache_;
+
+ //
+ // Get rid of the image decompression
+ // resources.
+ //
+
+ UnpackDestroy();
+
+ #ifdef DEBUG
+ *logofs << "Proxy: Deleted proxy object at " << this
+ << ".\n" << logofs_flush;
+ #endif
+}
+
+int Proxy::setOperational()
+{
+ #ifdef TEST
+ *logofs << "Proxy: Entering operational mode.\n"
+ << logofs_flush;
+ #endif
+
+ operation_ = operation_in_messages;
+
+ return 1;
+}
+
+int Proxy::setReadDescriptors(fd_set *fdSet, int &fdMax, T_timestamp &tsMax)
+{
+ //
+ // Set the initial timeout to the time of
+ // the next ping. If the congestion count
+ // is greater than zero, anyway, use a
+ // shorter timeout to force a congestion
+ // update.
+ //
+
+ if (agent_ != nothing && congestions_[agent_] == 0 &&
+ statistics -> getCongestionInFrame() >= 1 &&
+ tokens_[token_control].remaining >=
+ (tokens_[token_control].limit - 1))
+ {
+ setMinTimestamp(tsMax, control -> IdleTimeout);
+
+ #ifdef TEST
+ *logofs << "Proxy: Initial timeout is " << tsMax.tv_sec
+ << " S and " << (double) tsMax.tv_usec /
+ 1000 << " Ms with congestion "
+ << statistics -> getCongestionInFrame()
+ << ".\n" << logofs_flush;
+ #endif
+ }
+ else
+ {
+ setMinTimestamp(tsMax, control -> PingTimeout);
+
+ #ifdef TEST
+ *logofs << "Proxy: Initial timeout is " << tsMax.tv_sec
+ << " S and " << (double) tsMax.tv_usec /
+ 1000 << " Ms.\n" << logofs_flush;
+ #endif
+ }
+
+ int fd = -1;
+
+ if (isTimeToRead() == 1)
+ {
+ //
+ // If we don't have split tokens available
+ // don't set the timeout.
+ //
+
+ if (tokens_[token_split].remaining > 0 &&
+ isTimestamp(timeouts_.splitTs) == 1)
+ {
+ int diffTs = getTimeToNextSplit();
+
+ #if defined(TEST) || defined(INFO) || \
+ defined(FLUSH) || defined(SPLIT)
+
+ if (diffTimestamp(timeouts_.splitTs,
+ getTimestamp()) > timeouts_.split)
+ {
+ *logofs << "Proxy: FLUSH! SPLIT! WARNING! Running with "
+ << diffTimestamp(timeouts_.splitTs, getTimestamp())
+ << " Ms elapsed since the last split.\n"
+ << logofs_flush;
+ }
+
+ *logofs << "Proxy: FLUSH! SPLIT! Requesting timeout of "
+ << diffTs << " Ms as there are splits to send.\n"
+ << logofs_flush;
+
+ #endif
+
+ setMinTimestamp(tsMax, diffTs);
+ }
+ #if defined(TEST) || defined(INFO)
+ else if (isTimestamp(timeouts_.splitTs) == 1)
+ {
+ *logofs << "Proxy: WARNING! Not requesting a split "
+ << "timeout with " << tokens_[token_split].remaining
+ << " split tokens remaining.\n" << logofs_flush;
+ }
+ #endif
+
+ //
+ // Loop through the valid channels and set
+ // the descriptors selected for read and
+ // the timeout.
+ //
+
+ T_list &channelList = activeChannels_.getList();
+
+ for (T_list::iterator j = channelList.begin();
+ j != channelList.end(); j++)
+ {
+ int channelId = *j;
+
+ if (channels_[channelId] == NULL)
+ {
+ continue;
+ }
+
+ fd = getFd(channelId);
+
+ if (channels_[channelId] -> getFinish() == 0 &&
+ (channels_[channelId] -> getType() == channel_x11 ||
+ tokens_[token_data].remaining > 0) &&
+ congestions_[channelId] == 0)
+ {
+ FD_SET(fd, fdSet);
+
+ if (fd >= fdMax)
+ {
+ fdMax = fd + 1;
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Descriptor FD#" << fd
+ << " selected for read with buffer length "
+ << transports_[channelId] -> length()
+ << ".\n" << logofs_flush;
+ #endif
+
+ //
+ // Wakeup the proxy if there are motion
+ // events to flush.
+ //
+
+ if (isTimestamp(timeouts_.motionTs) == 1)
+ {
+ int diffTs = getTimeToNextMotion();
+
+ #if defined(TEST) || defined(INFO)
+
+ if (diffTimestamp(timeouts_.motionTs,
+ getTimestamp()) > timeouts_.motion)
+ {
+ *logofs << "Proxy: FLUSH! WARNING! Running with "
+ << diffTimestamp(timeouts_.motionTs, getTimestamp())
+ << " Ms elapsed since the last motion.\n"
+ << logofs_flush;
+ }
+
+ *logofs << "Proxy: FLUSH! Requesting timeout of "
+ << diffTs << " Ms as FD#" << fd << " has motion "
+ << "events to send.\n" << logofs_flush;
+
+ #endif
+
+ setMinTimestamp(tsMax, diffTs);
+ }
+ }
+ #if defined(TEST) || defined(INFO)
+ else
+ {
+ if (channels_[channelId] -> getType() != channel_x11 &&
+ tokens_[token_data].remaining <= 0)
+ {
+ *logofs << "Proxy: WARNING! Descriptor FD#" << fd
+ << " not selected for read with "
+ << tokens_[token_data].remaining << " data "
+ << "tokens remaining.\n" << logofs_flush;
+ }
+ }
+ #endif
+ }
+ }
+ #if defined(TEST) || defined(INFO)
+ else
+ {
+ *logofs << "Proxy: WARNING! Disabled reading from channels.\n"
+ << logofs_flush;
+
+ *logofs << "Proxy: WARNING! Congestion is " << congestion_
+ << " pending " << transport_ -> pending() << " blocked "
+ << transport_ -> blocked() << " length " << transport_ ->
+ length() << ".\n" << logofs_flush;
+ }
+ #endif
+
+ //
+ // Include the proxy descriptor.
+ //
+
+ FD_SET(fd_, fdSet);
+
+ if (fd_ >= fdMax)
+ {
+ fdMax = fd_ + 1;
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Proxy descriptor FD#" << fd_
+ << " selected for read with buffer length "
+ << transport_ -> length() << ".\n"
+ << logofs_flush;
+ #endif
+
+ return 1;
+}
+
+//
+// Add to the mask the file descriptors of all
+// X connections to write to.
+//
+
+int Proxy::setWriteDescriptors(fd_set *fdSet, int &fdMax, T_timestamp &tsMax)
+{
+ int fd = -1;
+
+ T_list &channelList = activeChannels_.getList();
+
+ for (T_list::iterator j = channelList.begin();
+ j != channelList.end(); j++)
+ {
+ int channelId = *j;
+
+ if (channels_[channelId] != NULL)
+ {
+ fd = getFd(channelId);
+
+ if (transports_[channelId] -> length() > 0)
+ {
+ FD_SET(fd, fdSet);
+
+ #ifdef TEST
+ *logofs << "Proxy: Descriptor FD#" << fd << " selected "
+ << "for write with blocked " << transports_[channelId] ->
+ blocked() << " and length " << transports_[channelId] ->
+ length() << ".\n" << logofs_flush;
+ #endif
+
+ if (fd >= fdMax)
+ {
+ fdMax = fd + 1;
+ }
+ }
+ #ifdef TEST
+ else
+ {
+ *logofs << "Proxy: Descriptor FD#" << fd << " not selected "
+ << "for write with blocked " << transports_[channelId] ->
+ blocked() << " and length " << transports_[channelId] ->
+ length() << ".\n" << logofs_flush;
+ }
+ #endif
+
+ #if defined(TEST) || defined(INFO)
+
+ if (transports_[channelId] -> getType() !=
+ transport_agent && transports_[channelId] ->
+ length() > 0 && transports_[channelId] ->
+ blocked() != 1)
+ {
+ *logofs << "Proxy: PANIC! Descriptor FD#" << fd
+ << " has data to write but blocked flag is "
+ << transports_[channelId] -> blocked()
+ << ".\n" << logofs_flush;
+
+ cerr << "Error" << ": Descriptor FD#" << fd
+ << " has data to write but blocked flag is "
+ << transports_[channelId] -> blocked()
+ << ".\n";
+
+ HandleCleanup();
+ }
+
+ #endif
+ }
+ }
+
+ //
+ // Check if the proxy transport has data
+ // from a previous blocking write.
+ //
+
+ if (transport_ -> blocked() == 1)
+ {
+ FD_SET(fd_, fdSet);
+
+ #ifdef TEST
+ *logofs << "Proxy: Proxy descriptor FD#"
+ << fd_ << " selected for write. Blocked is "
+ << transport_ -> blocked() << " length is "
+ << transport_ -> length() << ".\n"
+ << logofs_flush;
+ #endif
+
+ if (fd_ >= fdMax)
+ {
+ fdMax = fd_ + 1;
+ }
+ }
+ #ifdef TEST
+ else
+ {
+ *logofs << "Proxy: Proxy descriptor FD#"
+ << fd_ << " not selected for write. Blocked is "
+ << transport_ -> blocked() << " length is "
+ << transport_ -> length() << ".\n"
+ << logofs_flush;
+ }
+ #endif
+
+ //
+ // We are entering the main select. Save
+ // the timestamp of the last loop so that
+ // we can detect the clock drifts.
+ //
+
+ timeouts_.loopTs = getTimestamp();
+
+ return 1;
+}
+
+int Proxy::getChannels(T_channel_type type)
+{
+ int channels = 0;
+
+ T_list &channelList = activeChannels_.getList();
+
+ for (T_list::iterator j = channelList.begin();
+ j != channelList.end(); j++)
+ {
+ int channelId = *j;
+
+ if (channels_[channelId] != NULL &&
+ (type == channel_none ||
+ type == channels_[channelId] ->
+ getType()))
+ {
+ channels++;
+ }
+ }
+
+ return channels;
+}
+
+T_channel_type Proxy::getType(int fd)
+{
+ int channelId = getChannel(fd);
+
+ if (channelId < 0 || channels_[channelId] == NULL)
+ {
+ return channel_none;
+ }
+
+ return channels_[channelId] -> getType();
+}
+
+const char *Proxy::getTypeName(T_channel_type type)
+{
+ switch (type)
+ {
+ case channel_x11:
+ {
+ return "X";
+ }
+ case channel_cups:
+ {
+ return "CUPS";
+ }
+ case channel_smb:
+ {
+ return "SMB";
+ }
+ case channel_media:
+ {
+ return "media";
+ }
+ case channel_http:
+ {
+ return "HTTP";
+ }
+ case channel_font:
+ {
+ return "font";
+ }
+ case channel_slave:
+ {
+ return "slave";
+ }
+ default:
+ {
+ return "unknown";
+ }
+ }
+}
+
+const char *Proxy::getComputerName()
+{
+ //
+ // Strangely enough, under some Windows OSes SMB
+ // service doesn't bind to localhost. Fall back
+ // to localhost if can't find computer name in
+ // the environment. In future we should try to
+ // bind to localhost and then try the other IPs.
+ //
+
+ char *hostname = NULL;
+
+ #ifdef __CYGWIN32__
+
+ hostname = getenv("COMPUTERNAME");
+
+ #endif
+
+ if (hostname == NULL)
+ {
+ hostname = "localhost";
+ }
+
+ return hostname;
+}
+
+//
+// Handle data from channels selected for read.
+//
+
+int Proxy::handleRead(int &resultFds, fd_set &readSet)
+{
+ #ifdef DEBUG
+ *logofs << "Proxy: Checking descriptors selected for read.\n"
+ << logofs_flush;
+ #endif
+
+ T_list &channelList = activeChannels_.getList();
+
+ for (T_list::iterator j = channelList.begin();
+ j != channelList.end(); j++)
+ {
+ #ifdef DEBUG
+ *logofs << "Proxy: Looping with current channel "
+ << *j << ".\n" << logofs_flush;
+ #endif
+
+ int fd = getFd(*j);
+
+ if (fd >= 0 && resultFds > 0 && FD_ISSET(fd, &readSet))
+ {
+ #ifdef DEBUG
+ *logofs << "Proxy: Going to read messages from FD#"
+ << fd << ".\n" << logofs_flush;
+ #endif
+
+ int result = handleRead(fd);
+
+ if (result < 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Failure reading messages from FD#"
+ << fd << ".\n" << logofs_flush;
+ #endif
+
+ return -1;
+ }
+
+ #ifdef DEBUG
+ *logofs << "Proxy: Clearing the read descriptor "
+ << "for FD#" << fd << ".\n" << logofs_flush;
+ #endif
+
+ FD_CLR(fd, &readSet);
+
+ resultFds--;
+ }
+ }
+
+ if (resultFds > 0 && FD_ISSET(fd_, &readSet))
+ {
+ #ifdef DEBUG
+ *logofs << "Proxy: Going to read messages from "
+ << "proxy FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ if (handleRead() < 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Failure reading from proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ return -1;
+ }
+
+ #ifdef DEBUG
+ *logofs << "Proxy: Clearing the read descriptor "
+ << "for proxy FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ FD_CLR(fd_, &readSet);
+
+ resultFds--;
+ }
+
+ return 1;
+}
+
+//
+// Perform flush on descriptors selected for write.
+//
+
+int Proxy::handleFlush(int &resultFds, fd_set &writeSet)
+{
+ #ifdef DEBUG
+ *logofs << "Proxy: Checking descriptors selected for write.\n"
+ << logofs_flush;
+ #endif
+
+ if (resultFds > 0 && FD_ISSET(fd_, &writeSet))
+ {
+ #ifdef TEST
+ *logofs << "Proxy: FLUSH! Proxy descriptor FD#" << fd_
+ << " reported to be writable.\n"
+ << logofs_flush;
+ #endif
+
+ if (handleFlush() < 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Failure flushing the writable "
+ << "proxy FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ return -1;
+ }
+
+ #ifdef DEBUG
+ *logofs << "Proxy: Clearing the write descriptor "
+ << "for proxy FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ FD_CLR(fd_, &writeSet);
+
+ resultFds--;
+ }
+
+ T_list &channelList = activeChannels_.getList();
+
+ for (T_list::iterator j = channelList.begin();
+ resultFds > 0 && j != channelList.end(); j++)
+ {
+ #ifdef DEBUG
+ *logofs << "Proxy: Looping with current channel "
+ << *j << ".\n" << logofs_flush;
+ #endif
+
+ int fd = getFd(*j);
+
+ if (fd >= 0 && FD_ISSET(fd, &writeSet))
+ {
+ #ifdef TEST
+ *logofs << "Proxy: X descriptor FD#" << fd
+ << " reported to be writable.\n"
+ << logofs_flush;
+ #endif
+
+ //
+ // It can happen that, in handling reads, we
+ // have destroyed the buffer associated to a
+ // closed socket, so don't complain about
+ // the errors.
+ //
+
+ handleFlush(fd);
+
+ //
+ // Clear the descriptor from the mask so
+ // we don't confuse the agent if it's
+ // not checking only its own descriptors.
+ //
+
+ #ifdef DEBUG
+ *logofs << "Proxy: Clearing the write descriptor "
+ << "for FD#" << fd << ".\n"
+ << logofs_flush;
+ #endif
+
+ FD_CLR(fd, &writeSet);
+
+ resultFds--;
+ }
+ }
+
+ return 1;
+}
+
+int Proxy::handleRead()
+{
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Decoding data from proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ //
+ // Decode all the available messages from
+ // the remote proxy until is not possible
+ // to read more.
+ //
+
+ for (;;)
+ {
+ int result = readBuffer_.readMessage();
+
+ #if defined(TEST) || defined(DEBUG) || defined(INFO)
+ *logofs << "Proxy: Read result on proxy FD#" << fd_
+ << " is " << result << ".\n"
+ << logofs_flush;
+ #endif
+
+ if (result < 0)
+ {
+ if (shutdown_ == 0)
+ {
+ if (finish_ == 0)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Failure reading from the "
+ << "peer proxy on FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Failure reading from the "
+ << "peer proxy.\n";
+ }
+ }
+ #ifdef TEST
+ else
+ {
+ *logofs << "Proxy: Closure of the proxy link detected "
+ << "after clean shutdown.\n" << logofs_flush;
+ }
+ #endif
+
+ priority_ = 0;
+ finish_ = 1;
+ congestion_ = 0;
+
+ return -1;
+ }
+ else if (result == 0)
+ {
+ #if defined(TEST) || defined(DEBUG) || defined(INFO)
+ *logofs << "Proxy: No data read from proxy FD#"
+ << fd_ << "\n" << logofs_flush;
+ #endif
+
+ return 0;
+ }
+
+ //
+ // We read some data from the remote. If we set
+ // the congestion flag because we couldn't read
+ // before the timeout and have tokens available,
+ // then reset the congestion flag.
+ //
+
+ if (congestion_ == 1 &&
+ tokens_[token_control].remaining > 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Exiting congestion due to "
+ << "proxy data with " << tokens_[token_control].remaining
+ << " tokens.\n" << logofs_flush;
+ #endif
+
+ congestion_ = 0;
+ }
+
+ //
+ // Set the timestamp of the last read
+ // operation from the remote proxy and
+ // enable again showing the 'no data
+ // received' dialog at the next timeout.
+ //
+
+ timeouts_.readTs = getTimestamp();
+
+ if (alert_ != 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Displacing the dialog "
+ << "for proxy FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ HandleAlert(DISPLACE_MESSAGE_ALERT, 1);
+ }
+
+ timeouts_.alertTs = nullTimestamp();
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Getting messages from proxy FD#" << fd_
+ << " with " << readBuffer_.getLength() << " bytes "
+ << "in the read buffer.\n" << logofs_flush;
+ #endif
+
+ unsigned int controlLength;
+ unsigned int dataLength;
+
+ const unsigned char *message;
+
+ while ((message = readBuffer_.getMessage(controlLength, dataLength)) != NULL)
+ {
+ statistics -> addFrameIn();
+
+ if (controlLength == 3 && *message == 0 &&
+ *(message + 1) < code_last_tag)
+ {
+ if (handleControlFromProxy(message) < 0)
+ {
+ return -1;
+ }
+ }
+ else if (operation_ == operation_in_messages)
+ {
+ int channelId = inputChannel_;
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Identified message of " << dataLength
+ << " bytes for FD#" << getFd(channelId) << " channel ID#"
+ << channelId << ".\n" << logofs_flush;
+ #endif
+
+ if (channelId >= 0 && channelId < CONNECTIONS_LIMIT &&
+ channels_[channelId] != NULL)
+ {
+ int finish = channels_[channelId] -> getFinish();
+
+ #ifdef WARNING
+
+ if (finish == 1)
+ {
+ *logofs << "Proxy: WARNING! Handling data for finishing "
+ << "FD#" << getFd(channelId) << " channel ID#"
+ << channelId << ".\n" << logofs_flush;
+ }
+
+ #endif
+
+ //
+ // We need to decode all the data to preserve
+ // the consistency of the cache, so can't re-
+ // turn as soon as the first error is encount-
+ // ered. Check if this is the first time that
+ // the failure is detected.
+ //
+
+ int result = channels_[channelId] -> handleWrite(message, dataLength);
+
+ if (result < 0 && finish == 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Failed to write proxy data to FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << ".\n" << logofs_flush;
+ #endif
+
+ if (handleFinish(channelId) < 0)
+ {
+ return -1;
+ }
+ }
+
+ //
+ // Check if we have splits or motion
+ // events to send.
+ //
+
+ setSplitTimeout(channelId);
+ setMotionTimeout(channelId);
+ }
+ #ifdef WARNING
+ else
+ {
+ *logofs << "Proxy: WARNING! Received data for "
+ << "invalid channel ID#" << channelId
+ << ".\n" << logofs_flush;
+ }
+ #endif
+ }
+ else if (operation_ == operation_in_statistics)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Received statistics data from remote proxy.\n"
+ << logofs_flush;
+ #endif
+
+ if (handleStatisticsFromProxy(message, dataLength) < 0)
+ {
+ return -1;
+ }
+
+ operation_ = operation_in_messages;
+ }
+ else if (operation_ == operation_in_negotiation)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Received new negotiation data from remote proxy.\n"
+ << logofs_flush;
+ #endif
+
+ if (handleNegotiationFromProxy(message, dataLength) < 0)
+ {
+ return -1;
+ }
+ }
+
+ //
+ // if (controlLength == 3 && *message == 0 && ...) ...
+ // else if (operation_ == operation_in_statistics) ...
+ // else if (operation_ == operation_in_messages) ...
+ // else if (operation_ == operation_in_negotiation) ...
+ // else ...
+ //
+
+ else
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Unrecognized message received on proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Unrecognized message received on proxy FD#"
+ << fd_ << ".\n";
+
+ return -1;
+ }
+
+ } // while ((message = readBuffer_.getMessage(controlLength, dataLength)) != NULL) ...
+
+ //
+ // Reset the read buffer.
+ //
+
+ readBuffer_.fullReset();
+
+ //
+ // Give up if no data is readable.
+ //
+
+ if (transport_ -> readable() == 0)
+ {
+ break;
+ }
+
+ } // End of for (;;) ...
+
+ return 1;
+}
+
+int Proxy::handleControlFromProxy(const unsigned char *message)
+{
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Received message '" << DumpControl(*(message + 1))
+ << "' at " << strMsTimestamp() << " with data ID#"
+ << (int) *(message + 2) << ".\n" << logofs_flush;
+ #endif
+
+ T_channel_type channelType = channel_none;
+
+ switch (*(message + 1))
+ {
+ case code_switch_connection:
+ {
+ int channelId = *(message + 2);
+
+ //
+ // If channel is invalid further messages will
+ // be ignored. The acknowledged shutdown of
+ // channels should prevent this.
+ //
+
+ inputChannel_ = channelId;
+
+ break;
+ }
+ case code_begin_congestion:
+ {
+ //
+ // Set the congestion state for the
+ // channel reported by the remote.
+ //
+
+ int channelId = *(message + 2);
+
+ if (channels_[channelId] != NULL)
+ {
+ congestions_[channelId] = 1;
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Received a begin congestion "
+ << "for channel id ID#" << channelId
+ << ".\n" << logofs_flush;
+ #endif
+
+ if (channelId == agent_ && congestions_[agent_] != 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Forcing an update of the congestion "
+ << "counter with agent congested.\n"
+ << logofs_flush;
+ #endif
+
+ statistics -> updateCongestion(-tokens_[token_control].remaining,
+ tokens_[token_control].limit);
+ }
+ }
+ #ifdef WARNING
+ else
+ {
+ *logofs << "Proxy: WARNING! Received a begin congestion "
+ << "for invalid channel id ID#" << channelId
+ << ".\n" << logofs_flush;
+ }
+ #endif
+
+ break;
+ }
+ case code_end_congestion:
+ {
+ //
+ // Attend again to the channel.
+ //
+
+ int channelId = *(message + 2);
+
+ if (channels_[channelId] != NULL)
+ {
+ congestions_[channelId] = 0;
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Received an end congestion "
+ << "for channel id ID#" << channelId
+ << ".\n" << logofs_flush;
+ #endif
+
+ if (channelId == agent_ && congestions_[agent_] != 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Forcing an update of the congestion "
+ << "counter with agent decongested.\n"
+ << logofs_flush;
+ #endif
+
+ statistics -> updateCongestion(tokens_[token_control].remaining,
+ tokens_[token_control].limit);
+ }
+ }
+ #ifdef WARNING
+ else
+ {
+ *logofs << "Proxy: WARNING! Received an end congestion "
+ << "for invalid channel id ID#" << channelId
+ << ".\n" << logofs_flush;
+ }
+ #endif
+
+ break;
+ }
+ case code_control_token_request:
+ {
+ T_proxy_token &token = tokens_[token_control];
+
+ if (handleTokenFromProxy(token, *(message + 2)) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case code_split_token_request:
+ {
+ T_proxy_token &token = tokens_[token_split];
+
+ if (handleTokenFromProxy(token, *(message + 2)) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case code_data_token_request:
+ {
+ T_proxy_token &token = tokens_[token_data];
+
+ if (handleTokenFromProxy(token, *(message + 2)) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case code_control_token_reply:
+ {
+ T_proxy_token &token = tokens_[token_control];
+
+ if (handleTokenReplyFromProxy(token, *(message + 2)) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case code_split_token_reply:
+ {
+ T_proxy_token &token = tokens_[token_split];
+
+ if (handleTokenReplyFromProxy(token, *(message + 2)) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case code_data_token_reply:
+ {
+ T_proxy_token &token = tokens_[token_data];
+
+ if (handleTokenReplyFromProxy(token, *(message + 2)) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case code_new_x_connection:
+ {
+ //
+ // Opening the channel is handled later.
+ //
+
+ channelType = channel_x11;
+
+ break;
+ }
+ case code_new_cups_connection:
+ {
+ channelType = channel_cups;
+
+ break;
+ }
+ case code_new_aux_connection:
+ {
+ //
+ // Starting from version 1.5.0 we create real X
+ // connections for the keyboard channel. We need
+ // to refuse old auxiliary X connections because
+ // they would be unable to leverage the new fake
+ // authorization cookie.
+ //
+
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Can't open outdated auxiliary X "
+ << "channel for code " << *(message + 1) << ".\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Can't open outdated auxiliary X "
+ << "channel for code " << *(message + 1) << ".\n";
+
+ if (handleControl(code_drop_connection, *(message + 2)) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case code_new_smb_connection:
+ {
+ channelType = channel_smb;
+
+ break;
+ }
+ case code_new_media_connection:
+ {
+ channelType = channel_media;
+
+ break;
+ }
+ case code_new_http_connection:
+ {
+ channelType = channel_http;
+
+ break;
+ }
+ case code_new_font_connection:
+ {
+ channelType = channel_font;
+
+ break;
+ }
+ case code_new_slave_connection:
+ {
+ channelType = channel_slave;
+
+ break;
+ }
+ case code_drop_connection:
+ {
+ int channelId = *(message + 2);
+
+ if (channelId >= 0 && channelId < CONNECTIONS_LIMIT &&
+ channels_[channelId] != NULL)
+ {
+ handleDropFromProxy(channelId);
+ }
+ #ifdef WARNING
+ else
+ {
+ *logofs << "Proxy: WARNING! Received a drop message "
+ << "for invalid channel id ID#" << channelId
+ << ".\n" << logofs_flush;
+ }
+ #endif
+
+ break;
+ }
+ case code_finish_connection:
+ {
+ int channelId = *(message + 2);
+
+ if (channelId >= 0 && channelId < CONNECTIONS_LIMIT &&
+ channels_[channelId] != NULL)
+ {
+ //
+ // Force the finish state on the channel.
+ // We can receive this message while in
+ // the read loop, so we only mark the
+ // channel for deletion.
+ //
+
+ #ifdef TEST
+ *logofs << "Proxy: Received a finish message for FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << ".\n" << logofs_flush;
+ #endif
+
+ handleFinishFromProxy(channelId);
+ }
+ #ifdef WARNING
+ else
+ {
+ *logofs << "Proxy: WARNING! Received a finish message "
+ << "for invalid channel id ID#" << channelId
+ << ".\n" << logofs_flush;
+ }
+ #endif
+
+ break;
+ }
+ case code_finish_listeners:
+ {
+ //
+ // This is from the main loop.
+ //
+
+ #ifdef TEST
+ *logofs << "Proxy: Closing down all local listeners.\n"
+ << logofs_flush;
+ #endif
+
+ CleanupListeners();
+
+ finish_ = 1;
+
+ break;
+ }
+ case code_reset_request:
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Proxy reset not supported "
+ << "in this version.\n" << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Proxy reset not supported "
+ << "in this version.\n";
+
+ HandleCleanup();
+ }
+ case code_shutdown_request:
+ {
+ //
+ // Time to rest in peace.
+ //
+
+ shutdown_ = 1;
+
+ break;
+ }
+ case code_load_request:
+ {
+ if (handleLoadFromProxy() < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case code_save_request:
+ {
+ //
+ // Don't abort the connection
+ // if can't write to disk.
+ //
+
+ handleSaveFromProxy();
+
+ break;
+ }
+ case code_statistics_request:
+ {
+ int type = *(message + 2);
+
+ if (handleStatisticsFromProxy(type) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case code_statistics_reply:
+ {
+ operation_ = operation_in_statistics;
+
+ break;
+ }
+ case code_alert_request:
+ {
+ HandleAlert(*(message + 2), 1);
+
+ break;
+ }
+ case code_sync_request:
+ {
+ int channelId = *(message + 2);
+
+ if (handleSyncFromProxy(channelId) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case code_sync_reply:
+ {
+ //
+ // We are not the one that issued
+ // the request.
+ //
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: PANIC! Received an unexpected "
+ << "synchronization reply.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Received an unexpected "
+ << "synchronization reply.\n";
+
+ HandleCleanup();
+ }
+ default:
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Received bad control message number "
+ << (unsigned int) *(message + 1) << " with attribute "
+ << (unsigned int) *(message + 2) << ".\n" << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Received bad control message number "
+ << (unsigned int) *(message + 1) << " with attribute "
+ << (unsigned int) *(message + 2) << ".\n";
+
+ HandleCleanup();
+ }
+
+ } // End of switch (*(message + 1)) ...
+
+ if (channelType == channel_none)
+ {
+ return 1;
+ }
+
+ //
+ // Handle the channel allocation that we
+ // left from the main switch case.
+ //
+
+ int channelId = *(message + 2);
+
+ //
+ // Check if the channel is in the valid
+ // range.
+ //
+
+ int result = checkChannelMap(channelId);
+
+ if (result >= 0)
+ {
+ result = handleNewConnectionFromProxy(channelType, channelId);
+ }
+
+ if (result < 0)
+ {
+ //
+ // Realization of new channel failed.
+ // Send channel shutdown message to
+ // the peer proxy.
+ //
+
+ if (handleControl(code_drop_connection, channelId) < 0)
+ {
+ return -1;
+ }
+ }
+ else
+ {
+ int fd = getFd(channelId);
+
+ if (getReadable(fd) > 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Trying to read immediately "
+ << "from descriptor FD#" << fd << ".\n"
+ << logofs_flush;
+ #endif
+
+ if (handleRead(fd) < 0)
+ {
+ return -1;
+ }
+ }
+ #ifdef TEST
+ *logofs << "Proxy: Nothing to read immediately "
+ << "from descriptor FD#" << fd << ".\n"
+ << logofs_flush;
+ #endif
+ }
+
+ return 1;
+}
+
+int Proxy::handleRead(int fd, const char *data, int size)
+{
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Handling data for connection on FD#"
+ << fd << ".\n" << logofs_flush;
+ #endif
+
+ if (canRead(fd) == 0)
+ {
+ #if defined(TEST) || defined(INFO)
+
+ if (getChannel(fd) < 0)
+ {
+ *logofs << "Proxy: PANIC! Can't read from invalid FD#"
+ << fd << ".\n" << logofs_flush;
+
+ HandleCleanup();
+ }
+ else
+ {
+ *logofs << "Proxy: WARNING! Read method called for FD#"
+ << fd << " but operation is not possible.\n"
+ << logofs_flush;
+ }
+
+ #endif
+
+ return 0;
+ }
+
+ int channelId = getChannel(fd);
+
+ //
+ // Let the channel object read all the new data from
+ // its file descriptor, isolate messages, compress
+ // those messages, and append the compressed form to
+ // the encode buffer.
+ //
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Reading messages from FD#" << fd
+ << " channel ID#" << channelId << ".\n"
+ << logofs_flush;
+ #endif
+
+ int result = channels_[channelId] -> handleRead(encodeBuffer_, (const unsigned char *) data,
+ (unsigned int) size);
+
+ //
+ // Even in the case of a failure, write the produced
+ // data to the proxy connection. To keep the stores
+ // synchronized, the remote side needs to decode any
+ // message encoded by this side, also if the X socket
+ // was closed in the meanwhile. If this is the case,
+ // the decompressed output will be silently discarded.
+ //
+
+ if (result < 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Failed to read data from connection FD#"
+ << fd << " channel ID#" << channelId << ".\n"
+ << logofs_flush;
+ #endif
+
+ if (handleFinish(channelId) < 0)
+ {
+ return -1;
+ }
+ }
+
+ //
+ // Check if there are new splits or
+ // motion events to send.
+ //
+
+ setSplitTimeout(channelId);
+ setMotionTimeout(channelId);
+
+ return 1;
+}
+
+int Proxy::handleEvents()
+{
+ #ifdef TEST
+ *logofs << "Proxy: Going to check the events on channels.\n"
+ << logofs_flush;
+ #endif
+
+ //
+ // Check if we can safely write to the
+ // proxy link.
+ //
+
+ int read = isTimeToRead();
+
+ //
+ // Loop on channels and send the pending
+ // events. We must copy the list because
+ // channels can be removed in the middle
+ // of the loop.
+ //
+
+ T_list channelList = activeChannels_.copyList();
+
+ for (T_list::iterator j = channelList.begin();
+ j != channelList.end(); j++)
+ {
+ int channelId = *j;
+
+ if (channels_[channelId] == NULL)
+ {
+ continue;
+ }
+
+ //
+ // Check if we need to drop the channel.
+ //
+
+ if (channels_[channelId] -> getDrop() == 1 ||
+ channels_[channelId] -> getClosing() == 1)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Dropping the descriptor FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << ".\n" << logofs_flush;
+ #endif
+
+ if (handleDrop(channelId) < 0)
+ {
+ return -1;
+ }
+
+ continue;
+ }
+ else if (channels_[channelId] -> getFinish() == 1)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Skipping finishing "
+ << "descriptor FD#" << getFd(channelId)
+ << " channel ID#" << channelId << ".\n"
+ << logofs_flush;
+ #endif
+
+ continue;
+ }
+
+ //
+ // If the proxy link or the channel is
+ // in congestion state, don't handle
+ // the further events.
+ //
+
+ if (read == 0 || congestions_[channelId] == 1)
+ {
+ #ifdef TEST
+
+ if (read == 0)
+ {
+ *logofs << "Proxy: Can't handle events for FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << " with proxy not available.\n"
+ << logofs_flush;
+ }
+ else
+ {
+ *logofs << "Proxy: Can't handle events for FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << " with channel congested.\n"
+ << logofs_flush;
+ }
+
+ #endif
+
+ continue;
+ }
+
+ //
+ // Handle the timeouts on the channel
+ // operations.
+ //
+
+ int result = 0;
+
+ //
+ // Handle the motion events.
+ //
+
+ if (result >= 0 && channels_[channelId] -> needMotion() == 1)
+ {
+ if (isTimeToMotion() == 1)
+ {
+ #if defined(TEST) || defined(INFO) || defined(FLUSH)
+
+ *logofs << "Proxy: FLUSH! Motion timeout expired after "
+ << diffTimestamp(timeouts_.motionTs, getTimestamp())
+ << " Ms.\n" << logofs_flush;
+
+ #endif
+
+ result = channels_[channelId] -> handleMotion(encodeBuffer_);
+
+ #ifdef TEST
+
+ if (result < 0)
+ {
+ *logofs << "Proxy: Failed to handle motion events for FD#"
+ << getFd(channelId) << " channel ID#" << channelId
+ << ".\n" << logofs_flush;
+ }
+
+ #endif
+
+ timeouts_.motionTs = nullTimestamp();
+
+ setMotionTimeout(channelId);
+ }
+ #if defined(TEST) || defined(INFO)
+ else if (isTimestamp(timeouts_.motionTs) == 1)
+ {
+ *logofs << "Proxy: Running with "
+ << diffTimestamp(timeouts_.motionTs, getTimestamp())
+ << " Ms elapsed since the last motion.\n"
+ << logofs_flush;
+ }
+ #endif
+ }
+
+ if (result >= 0 && channels_[channelId] -> needSplit() == 1)
+ {
+ //
+ // Check if it is time to send more splits
+ // and how many bytes are going to be sent.
+ //
+
+ if (isTimeToSplit() == 1)
+ {
+ #if defined(TEST) || defined(INFO) || defined(SPLIT)
+ *logofs << "Proxy: SPLIT! Split timeout expired after "
+ << diffTimestamp(timeouts_.splitTs, getTimestamp())
+ << " Ms.\n" << logofs_flush;
+ #endif
+
+ #if defined(TEST) || defined(INFO) || defined(SPLIT)
+
+ *logofs << "Proxy: SPLIT! Encoding splits for FD#"
+ << getFd(channelId) << " at " << strMsTimestamp()
+ << " with " << clientStore_ -> getSplitTotalStorageSize()
+ << " total bytes and " << control -> SplitDataPacketLimit
+ << " bytes " << "to write.\n"
+ << logofs_flush;
+
+ #endif
+
+ result = channels_[channelId] -> handleSplit(encodeBuffer_);
+
+ #ifdef TEST
+
+ if (result < 0)
+ {
+ *logofs << "Proxy: Failed to handle splits for FD#"
+ << getFd(channelId) << " channel ID#" << channelId
+ << ".\n" << logofs_flush;
+ }
+
+ #endif
+
+ timeouts_.splitTs = nullTimestamp();
+
+ setSplitTimeout(channelId);
+ }
+ #if defined(TEST) || defined(INFO) || defined(SPLIT)
+ else if (channels_[channelId] -> needSplit() == 1 &&
+ isTimestamp(timeouts_.splitTs) == 0)
+ {
+ *logofs << "Proxy: SPLIT! WARNING! Channel for FD#"
+ << getFd(channelId) << " has split to send but "
+ << "there is no timeout.\n" << logofs_flush;
+ }
+ else if (isTimestamp(timeouts_.splitTs) == 1)
+ {
+ *logofs << "Proxy: SPLIT! Running with "
+ << diffTimestamp(timeouts_.splitTs, getTimestamp())
+ << " Ms elapsed since the last split.\n"
+ << logofs_flush;
+ }
+ #endif
+ }
+
+ if (result < 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Error handling events for FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << ".\n" << logofs_flush;
+ #endif
+
+ if (handleFinish(channelId) < 0)
+ {
+ return -1;
+ }
+ }
+ }
+
+ return 1;
+}
+
+int Proxy::handleFrame(T_frame_type type)
+{
+ //
+ // Write any outstanding control message, followed by the
+ // content of the encode buffer, to the proxy transport.
+ //
+ // This code assumes that the encode buffer data is at an
+ // offset several bytes from start of the buffer, so that
+ // the length header and any necessary control bytes can
+ // be inserted in front of the data already in the buffer.
+ // This is the easiest way to encapsulate header and data
+ // together in a single frame.
+ //
+ // The way framing is implemented is very efficient but
+ // inherently limited and does not allow for getting the
+ // best performance, especially when running over a fast
+ // link. Framing should be rewritten to include the length
+ // of the packets in a fixed size header and, possibly,
+ // to incapsulate the control messages and the channel's
+ // data in a pseudo X protocol message, so that the proxy
+ // itself would be treated like any other channel.
+ //
+
+ #if defined(TEST) || defined(INFO)
+
+ if (congestion_ == 1)
+ {
+ //
+ // This can happen because there may be control
+ // messages to send, like a proxy shutdown mes-
+ // sage or a statistics request. All the other
+ // cases should be considered an error.
+ //
+
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Data is to be sent while "
+ << "congestion is " << congestion_ << ".\n"
+ << logofs_flush;
+ #endif
+ }
+
+ #endif
+
+ //
+ // Check if there is any data available on
+ // the socket. Recent Linux kernels are very
+ // picky. They require that we read often or
+ // they assume that the process is non-inter-
+ // active.
+ //
+
+ if (handleAsyncEvents() < 0)
+ {
+ return -1;
+ }
+
+ //
+ // Check if this is a ping, not a data frame.
+ //
+
+ if (type == frame_ping)
+ {
+ if (handleToken(frame_ping) < 0)
+ {
+ return -1;
+ }
+ }
+
+ unsigned int dataLength = encodeBuffer_.getLength();
+
+ #ifdef DEBUG
+ *logofs << "Proxy: Data length is " << dataLength
+ << " control length is " << controlLength_
+ << ".\n" << logofs_flush;
+ #endif
+
+ if (dataLength > 0)
+ {
+ //
+ // If this is a generic channel we need
+ // to add the completion bits. Data can
+ // also have been encoded because of a
+ // statistics request, even if no output
+ // channel was currently selected.
+ //
+
+ if (outputChannel_ != -1)
+ {
+ #if defined(TEST) || defined(INFO)
+
+ if (channels_[outputChannel_] == NULL)
+ {
+ *logofs << "Proxy: PANIC! A new frame was requested "
+ << "but the channel is invalid.\n"
+ << logofs_flush;
+
+ HandleCleanup();
+ }
+
+ #endif
+
+ channels_[outputChannel_] -> handleCompletion(encodeBuffer_);
+
+ dataLength = encodeBuffer_.getLength();
+ }
+ }
+ else if (controlLength_ == 0)
+ {
+ #if defined(TEST) || defined(INFO)
+
+ *logofs << "Proxy: PANIC! A new frame was requested "
+ << "but there is no data to write.\n"
+ << logofs_flush;
+
+ HandleCleanup();
+
+ #endif
+
+ return 0;
+ }
+
+ #ifdef DEBUG
+ *logofs << "Proxy: Data length is now " << dataLength
+ << " control length is " << controlLength_
+ << ".\n" << logofs_flush;
+ #endif
+
+ //
+ // Check if this frame needs to carry a new
+ // token request.
+ //
+
+ if (type == frame_data)
+ {
+ if (handleToken(frame_data) < 0)
+ {
+ return -1;
+ }
+ }
+
+ #ifdef DEBUG
+ *logofs << "Proxy: Adding a new frame for the remote proxy.\n"
+ << logofs_flush;
+ #endif
+
+ unsigned char temp[5];
+
+ unsigned int lengthLength = 0;
+ unsigned int shift = dataLength;
+
+ while (shift)
+ {
+ temp[lengthLength++] = (unsigned char) (shift & 0x7f);
+
+ shift >>= 7;
+ }
+
+ unsigned char *data = encodeBuffer_.getData();
+
+ unsigned char *outputMessage = data - (controlLength_ + lengthLength);
+
+ unsigned char *nextDest = outputMessage;
+
+ for (int i = 0; i < controlLength_; i++)
+ {
+ *nextDest++ = controlCodes_[i];
+ }
+
+ for (int j = lengthLength - 1; j > 0; j--)
+ {
+ *nextDest++ = (temp[j] | 0x80);
+ }
+
+ if (lengthLength)
+ {
+ *nextDest++ = temp[0];
+ }
+
+ unsigned int outputLength = dataLength + controlLength_ + lengthLength;
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Produced plain output for " << dataLength << "+"
+ << controlLength_ << "+" << lengthLength << " out of "
+ << outputLength << " bytes.\n" << logofs_flush;
+ #endif
+
+ #if defined(TEST) || defined(INFO) || defined(FLUSH) || defined(TIME)
+
+ T_timestamp nowTs = getTimestamp();
+
+ *logofs << "Proxy: FLUSH! Immediate with blocked " << transport_ ->
+ blocked() << " length " << transport_ -> length()
+ << " new " << outputLength << " flushable " << transport_ ->
+ flushable() << " tokens " << tokens_[token_control].remaining
+ << " after " << diffTimestamp(timeouts_.writeTs, nowTs)
+ << " Ms.\n" << logofs_flush;
+
+ *logofs << "Proxy: FLUSH! Immediate flush to proxy FD#" << fd_
+ << " of " << outputLength << " bytes at " << strMsTimestamp()
+ << " with priority " << priority_ << ".\n" << logofs_flush;
+
+ *logofs << "Proxy: FLUSH! Current bitrate is "
+ << statistics -> getBitrateInShortFrame() << " with "
+ << statistics -> getBitrateInLongFrame() << " in the "
+ << "long frame and top " << statistics ->
+ getTopBitrate() << ".\n" << logofs_flush;
+ #endif
+
+ statistics -> addWriteOut();
+
+ int result = transport_ -> write(write_immediate, outputMessage, outputLength);
+
+ #ifdef TIME
+
+ if (diffTimestamp(timeouts_.writeTs, nowTs) > 50)
+ {
+ *logofs << "Proxy: WARNING! TIME! Data written to proxy FD#"
+ << fd_ << " at " << strMsTimestamp() << " after "
+ << diffTimestamp(timeouts_.writeTs, nowTs)
+ << " Ms.\n" << logofs_flush;
+ }
+
+ #endif
+
+ #ifdef DUMP
+ *logofs << "Proxy: Sent " << outputLength << " bytes of data "
+ << "with checksum ";
+
+ DumpChecksum(outputMessage, outputLength);
+
+ *logofs << " on proxy FD#" << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ #ifdef DUMP
+ *logofs << "Proxy: Partial checksums are:\n";
+
+ DumpBlockChecksums(outputMessage, outputLength, 256);
+
+ *logofs << logofs_flush;
+ #endif
+
+ //
+ // Clean up the encode buffer and
+ // bring it to the initial size.
+ //
+
+ encodeBuffer_.fullReset();
+
+ //
+ // Close the connection if we got
+ // an error.
+ //
+
+ if (result < 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Failed write to proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ return -1;
+ }
+
+ //
+ // Account for the data frame and the
+ // framing overhead.
+ //
+
+ if (dataLength > 0)
+ {
+ statistics -> addFrameOut();
+ }
+
+ statistics -> addFramingBits((controlLength_ + lengthLength) << 3);
+
+ controlLength_ = 0;
+
+ //
+ // Reset all buffers, counters and the
+ // priority flag.
+ //
+
+ handleResetFlush();
+
+ //
+ // Check if more data became available
+ // after writing.
+ //
+
+ if (handleAsyncEvents() < 0)
+ {
+ return -1;
+ }
+
+ //
+ // Drain the proxy link if we are in
+ // congestion state.
+ //
+ // if (needDrain() == 1 && draining_ == 0)
+ // {
+ // if (handleDrain() < 0)
+ // {
+ // return -1;
+ // }
+ // }
+ //
+
+ return result;
+}
+
+int Proxy::handleFlush()
+{
+ //
+ // We can have data in the encode buffer or
+ // control bytes to send. In the case make
+ // up a new frame.
+ //
+
+ if (encodeBuffer_.getLength() + controlLength_ > 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Flushing data in the encode buffer.\n"
+ << logofs_flush;
+ #endif
+
+ priority_ = 1;
+
+ if (handleFrame(frame_data) < 0)
+ {
+ return -1;
+ }
+ }
+
+ //
+ // Check if we have something to write.
+ //
+
+ if (transport_ -> length() + transport_ -> flushable() == 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Nothing else to flush for proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ return 0;
+ }
+
+ #if defined(TEST) || defined(INFO)
+
+ if (transport_ -> blocked() == 0)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Proxy descriptor FD#" << fd_
+ << " has data to flush but the transport "
+ << "is not blocked.\n" << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Proxy descriptor FD#" << fd_
+ << " has data to flush but the transport "
+ << "is not blocked.\n";
+
+ HandleCleanup();
+ }
+
+ #endif
+
+ #if defined(TEST) || defined(INFO) || defined(FLUSH)
+ *logofs << "Proxy: FLUSH! Deferred with blocked " << transport_ ->
+ blocked() << " length " << transport_ -> length()
+ << " flushable " << transport_ -> flushable() << " tokens "
+ << tokens_[token_control].remaining << ".\n"
+ << logofs_flush;
+
+ *logofs << "Proxy: FLUSH! Deferred flush to proxy FD#" << fd_
+ << " of " << transport_ -> length() + transport_ ->
+ flushable() << " bytes at " << strMsTimestamp()
+ << " with priority " << priority_ << ".\n"
+ << logofs_flush;
+
+ *logofs << "Proxy: FLUSH! Current bitrate is "
+ << statistics -> getBitrateInShortFrame() << " with "
+ << statistics -> getBitrateInLongFrame() << " in the "
+ << "long frame and top " << statistics ->
+ getTopBitrate() << ".\n" << logofs_flush;
+ #endif
+
+ statistics -> addWriteOut();
+
+ int result = transport_ -> flush();
+
+ if (result < 0)
+ {
+ return -1;
+ }
+
+ //
+ // Reset the counters and update the
+ // timestamp of the last write.
+ //
+
+ handleResetFlush();
+
+ return result;
+}
+
+int Proxy::handleDrain()
+{
+ //
+ // If the proxy is run in the same process
+ // as SSH, we can't block or the program
+ // would not have a chance to read or write
+ // its data.
+ //
+
+ if (control -> LinkEncrypted == 1)
+ {
+ return 0;
+ }
+
+ if (needDrain() == 0 || draining_ == 1)
+ {
+ #if defined(TEST) || defined(INFO)
+
+ if (draining_ == 1)
+ {
+ *logofs << "Proxy: WARNING! Already draining proxy FD#"
+ << fd_ << " at " << strMsTimestamp() << ".\n"
+ << logofs_flush;
+ }
+ else
+ {
+ *logofs << "Proxy: WARNING! No need to drain proxy FD#"
+ << fd_ << " with congestion " << congestion_
+ << " length " << transport_ -> length()
+ << " and blocked " << transport_ -> blocked()
+ << ".\n" << logofs_flush;
+ }
+
+ #endif
+
+ return 0;
+ }
+
+ draining_ = 1;
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Going to drain the proxy FD#" << fd_
+ << " at " << strMsTimestamp() << ".\n"
+ << logofs_flush;
+ #endif
+
+ int timeout = control -> PingTimeout / 2;
+
+ T_timestamp startTs = getNewTimestamp();
+
+ T_timestamp nowTs = startTs;
+
+ int remaining;
+ int result;
+
+ //
+ // Keep draining the proxy socket while
+ // reading the incoming messages until
+ // the timeout is expired.
+ //
+
+ for (;;)
+ {
+ remaining = timeout - diffTimestamp(startTs, nowTs);
+
+ if (remaining <= 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Timeout raised while draining "
+ << "FD#" << fd_ << " at " << strMsTimestamp()
+ << " after " << diffTimestamp(startTs, nowTs)
+ << " Ms.\n" << logofs_flush;
+ #endif
+
+ result = 0;
+
+ goto ProxyDrainEnd;
+ }
+
+ if (transport_ -> length() > 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Trying to write to FD#" << fd_
+ << " at " << strMsTimestamp() << " with length "
+ << transport_ -> length() << " and "
+ << remaining << " Ms remaining.\n"
+ << logofs_flush;
+ #endif
+
+ result = transport_ -> drain(0, remaining);
+
+ if (result == -1)
+ {
+ result = -1;
+
+ goto ProxyDrainEnd;
+ }
+ else if (result == 0 && transport_ -> readable() > 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Decoding more data from proxy FD#"
+ << fd_ << " at " << strMsTimestamp() << " with "
+ << transport_ -> length() << " bytes to write and "
+ << transport_ -> readable() << " readable.\n"
+ << logofs_flush;
+ #endif
+
+ if (handleRead() < 0)
+ {
+ result = -1;
+
+ goto ProxyDrainEnd;
+ }
+ }
+ #if defined(TEST) || defined(INFO)
+ else if (result == 1)
+ {
+ *logofs << "Proxy: Transport for proxy FD#" << fd_
+ << " drained down to " << transport_ -> length()
+ << " bytes.\n" << logofs_flush;
+ }
+ #endif
+ }
+ else
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Waiting for more data from proxy "
+ << "FD#" << fd_ << " at " << strMsTimestamp()
+ << " with " << remaining << " Ms remaining.\n"
+ << logofs_flush;
+ #endif
+
+
+ result = transport_ -> wait(remaining);
+
+ if (result == -1)
+ {
+ result = -1;
+
+ goto ProxyDrainEnd;
+ }
+ else if (result > 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Decoding more data from proxy FD#"
+ << fd_ << " at " << strMsTimestamp() << " with "
+ << transport_ -> readable() << " bytes readable.\n"
+ << logofs_flush;
+ #endif
+
+ if (handleRead() < 0)
+ {
+ result = -1;
+
+ goto ProxyDrainEnd;
+ }
+ }
+ }
+
+ //
+ // Check if we finally got the tokens
+ // that would allow us to come out of
+ // the congestion state.
+ //
+
+ if (needDrain() == 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Got decongestion for proxy FD#"
+ << fd_ << " at " << strMsTimestamp() << " after "
+ << diffTimestamp(startTs, getTimestamp())
+ << " Ms.\n" << logofs_flush;
+ #endif
+
+ result = 1;
+
+ goto ProxyDrainEnd;
+ }
+
+ nowTs = getNewTimestamp();
+ }
+
+ProxyDrainEnd:
+
+ draining_ = 0;
+
+ return result;
+}
+
+int Proxy::handleFlush(int fd)
+{
+ int channelId = getChannel(fd);
+
+ if (channelId < 0 || channels_[channelId] == NULL)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: WARNING! Skipping flush on invalid "
+ << "descriptor FD#" << fd << " channel ID#"
+ << channelId << ".\n" << logofs_flush;
+ #endif
+
+ return 0;
+ }
+ else if (channels_[channelId] -> getFinish() == 1)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Skipping flush on finishing "
+ << "descriptor FD#" << fd << " channel ID#"
+ << channelId << ".\n" << logofs_flush;
+ #endif
+
+ return 0;
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Going to flush FD#" << fd
+ << " with blocked " << transports_[channelId] -> blocked()
+ << " length " << transports_[channelId] -> length()
+ << ".\n" << logofs_flush;
+ #endif
+
+ if (channels_[channelId] -> handleFlush() < 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Failed to flush data to FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << ".\n" << logofs_flush;
+ #endif
+
+ handleFinish(channelId);
+
+ return -1;
+ }
+
+ return 1;
+}
+
+int Proxy::handleStatistics(int type, ostream *stream)
+{
+ if (stream == NULL || control -> EnableStatistics == 0)
+ {
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Cannot produce statistics "
+ << " for proxy FD#" << fd_ << ". Invalid settings "
+ << "for statistics or stream.\n" << logofs_flush;
+ #endif
+
+ return 0;
+ }
+ else if (currentStatistics_ != NULL)
+ {
+ //
+ // Need to update the stream pointer as the
+ // previous one could have been destroyed.
+ //
+
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Replacing stream while producing "
+ << "statistics in stream at " << currentStatistics_
+ << " for proxy FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+ }
+
+ currentStatistics_ = stream;
+
+ //
+ // Get statistics of remote peer.
+ //
+
+ if (handleControl(code_statistics_request, type) < 0)
+ {
+ return -1;
+ }
+
+ return 1;
+}
+
+int Proxy::handleStatisticsFromProxy(int type)
+{
+ if (needFlush() == 1)
+ {
+ #if defined(TEST) || defined(INFO) || defined(FLUSH)
+ *logofs << "Proxy: WARNING! Data for the previous "
+ << "channel ID#" << outputChannel_
+ << " flushed in statistics.\n"
+ << logofs_flush;
+ #endif
+
+ if (handleFrame(frame_data) < 0)
+ {
+ return -1;
+ }
+ }
+
+ if (control -> EnableStatistics == 1)
+ {
+ //
+ // Allocate a buffer for the output.
+ //
+
+ char *buffer = new char[STATISTICS_LENGTH];
+
+ *buffer = '\0';
+
+ if (control -> ProxyMode == proxy_client)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Producing "
+ << (type == TOTAL_STATS ? "total" : "partial")
+ << " client statistics for proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ statistics -> getClientProtocolStats(type, buffer);
+
+ statistics -> getClientOverallStats(type, buffer);
+ }
+ else
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Producing "
+ << (type == TOTAL_STATS ? "total" : "partial")
+ << " server statistics for proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ statistics -> getServerProtocolStats(type, buffer);
+ }
+
+ if (type == PARTIAL_STATS)
+ {
+ statistics -> resetPartialStats();
+ }
+
+ unsigned int length = strlen((char *) buffer) + 1;
+
+ encodeBuffer_.encodeValue(type, 8);
+
+ encodeBuffer_.encodeValue(length, 32);
+
+ #ifdef TEST
+ *logofs << "Proxy: Encoding " << length
+ << " bytes of statistics data for proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ encodeBuffer_.encodeMemory((unsigned char *) buffer, length);
+
+ //
+ // Account statistics data as framing bits.
+ //
+
+ statistics -> addFramingBits(length << 3);
+
+ delete [] buffer;
+ }
+ else
+ {
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Got statistics request "
+ << "but local statistics are disabled.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Got statistics request "
+ << "but local statistics are disabled.\n";
+
+ type = NO_STATS;
+
+ encodeBuffer_.encodeValue(type, 8);
+
+ #ifdef TEST
+ *logofs << "Proxy: Sending error code to remote proxy on FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+ }
+
+ //
+ // The next write will flush the statistics
+ // data and the control message.
+ //
+
+ if (handleControl(code_statistics_reply, type) < 0)
+ {
+ return -1;
+ }
+
+ return 1;
+}
+
+int Proxy::handleStatisticsFromProxy(const unsigned char *message, unsigned int length)
+{
+ if (currentStatistics_ == NULL)
+ {
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Unexpected statistics data received "
+ << "from remote proxy on FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Unexpected statistics data received "
+ << "from remote proxy.\n";
+
+ return 0;
+ }
+
+ //
+ // Allocate the decode buffer and at least
+ // the 'type' field to see if there was an
+ // error.
+ //
+
+ DecodeBuffer decodeBuffer(message, length);
+
+ unsigned int type;
+
+ decodeBuffer.decodeValue(type, 8);
+
+ if (type == NO_STATS)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Couldn't get statistics from remote "
+ << "proxy on FD#" << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Couldn't get statistics from remote proxy.\n";
+ }
+ else if (type != TOTAL_STATS && type != PARTIAL_STATS)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Cannot produce statistics "
+ << "with qualifier '" << type << "'.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Cannot produce statistics "
+ << "with qualifier '" << type << "'.\n";
+
+ return -1;
+ }
+ else
+ {
+ unsigned int size;
+
+ decodeBuffer.decodeValue(size, 32);
+
+ char *buffer = new char[STATISTICS_LENGTH];
+
+ *buffer = '\0';
+
+ if (control -> EnableStatistics == 1)
+ {
+ if (control -> ProxyMode == proxy_client)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Finalizing "
+ << (type == TOTAL_STATS ? "total" : "partial")
+ << " client statistics for proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ statistics -> getClientCacheStats(type, buffer);
+
+ #ifdef TEST
+ *logofs << "Proxy: Decoding " << size
+ << " bytes of statistics data for proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ strncat(buffer, (char *) decodeBuffer.decodeMemory(size), size);
+
+ statistics -> getClientProtocolStats(type, buffer);
+
+ statistics -> getClientOverallStats(type, buffer);
+ }
+ else
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Finalizing "
+ << (type == TOTAL_STATS ? "total" : "partial")
+ << " server statistics for proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ statistics -> getServerCacheStats(type, buffer);
+
+ statistics -> getServerProtocolStats(type, buffer);
+
+ #ifdef TEST
+ *logofs << "Proxy: Decoding " << size
+ << " bytes of statistics data for proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ strncat(buffer, (char *) decodeBuffer.decodeMemory(size), size);
+ }
+
+ if (type == PARTIAL_STATS)
+ {
+ statistics -> resetPartialStats();
+ }
+
+ *currentStatistics_ << buffer;
+
+ //
+ // Mark the end of text to help external parsing.
+ //
+
+ *currentStatistics_ << '\4';
+
+ *currentStatistics_ << flush;
+ }
+ else
+ {
+ //
+ // It can be that statistics were enabled at the time
+ // we issued the request (otherwise we could not have
+ // set the stream), but now they have been disabled
+ // by user. We must decode statistics data if we want
+ // to keep the connection.
+ //
+
+ #ifdef TEST
+ *logofs << "Proxy: Discarding " << size
+ << " bytes of statistics data for proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ strncat(buffer, (char *) decodeBuffer.decodeMemory(size), size);
+ }
+
+ delete [] buffer;
+ }
+
+ currentStatistics_ = NULL;
+
+ return 1;
+}
+
+int Proxy::handleNegotiation(const unsigned char *message, unsigned int length)
+{
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Writing data during proxy "
+ << "negotiation is not implemented.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Writing data during proxy "
+ << "negotiation is not implemented.\n";
+
+ return -1;
+}
+
+int Proxy::handleNegotiationFromProxy(const unsigned char *message, unsigned int length)
+{
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Reading data during proxy "
+ << "negotiation is not implemented.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Reading data during proxy "
+ << "negotiation is not implemented.\n";
+
+ return -1;
+}
+
+int Proxy::handleAlert(int alert)
+{
+ if (handleControl(code_alert_request, alert) < 0)
+ {
+ return -1;
+ }
+
+ return 1;
+}
+
+int Proxy::handleCloseConnection(int clientFd)
+{
+ int channelId = getChannel(clientFd);
+
+ if (channels_[channelId] != NULL &&
+ channels_[channelId] -> getFinish() == 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Closing down the channel for FD#"
+ << clientFd << ".\n" << logofs_flush;
+ #endif
+
+ if (handleFinish(channelId) < 0)
+ {
+ return -1;
+ }
+
+ return 1;
+ }
+
+ return 0;
+}
+
+int Proxy::handleCloseAllXConnections()
+{
+ #ifdef TEST
+ *logofs << "Proxy: Closing down any remaining X channel.\n"
+ << logofs_flush;
+ #endif
+
+ T_list &channelList = activeChannels_.getList();
+
+ for (T_list::iterator j = channelList.begin();
+ j != channelList.end(); j++)
+ {
+ int channelId = *j;
+
+ if (channels_[channelId] != NULL &&
+ channels_[channelId] -> getType() == channel_x11 &&
+ channels_[channelId] -> getFinish() == 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Closing down the channel for FD#"
+ << getFd(channelId) << ".\n" << logofs_flush;
+ #endif
+
+ if (handleFinish(channelId) < 0)
+ {
+ return -1;
+ }
+ }
+ }
+
+ return 1;
+}
+
+int Proxy::handleCloseAllListeners()
+{
+ if (control -> isProtoStep7() == 1)
+ {
+ if (finish_ == 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Closing down all remote listeners.\n"
+ << logofs_flush;
+ #endif
+
+ if (handleControl(code_finish_listeners) < 0)
+ {
+ return -1;
+ }
+
+ finish_ = 1;
+ }
+ }
+ else
+ {
+ #ifdef TEST
+ *logofs << "Proxy: WARNING! Not sending unsupported "
+ << "'code_finish_listeners' message.\n"
+ << logofs_flush;
+ #endif
+
+ finish_ = 1;
+ }
+
+ return 1;
+}
+
+void Proxy::handleResetAlert()
+{
+ if (alert_ != 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: The proxy alert '" << alert_
+ << "' was displaced.\n" << logofs_flush;
+ #endif
+
+ alert_ = 0;
+ }
+
+ T_list &channelList = activeChannels_.getList();
+
+ for (T_list::iterator j = channelList.begin();
+ j != channelList.end(); j++)
+ {
+ int channelId = *j;
+
+ if (channels_[channelId] != NULL)
+ {
+ channels_[channelId] -> handleResetAlert();
+ }
+ }
+}
+
+int Proxy::handleFinish(int channelId)
+{
+ //
+ // Send any outstanding encoded data and
+ // do any finalization needed on the
+ // channel.
+ //
+
+ if (needFlush(channelId) == 1)
+ {
+ if (channels_[channelId] -> getFinish() == 1)
+ {
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! The finishing channel ID#"
+ << channelId << " has data to flush.\n"
+ << logofs_flush;
+ #endif
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(FLUSH)
+ *logofs << "Proxy: WARNING! Flushing data for the "
+ << "finishing channel ID#" << channelId
+ << ".\n" << logofs_flush;
+ #endif
+
+ if (handleFrame(frame_data) < 0)
+ {
+ return -1;
+ }
+ }
+
+ //
+ // Reset the congestion state and the
+ // timeouts, if needed.
+ //
+
+ congestions_[channelId] = 0;
+
+ setSplitTimeout(channelId);
+ setMotionTimeout(channelId);
+
+ if (channels_[channelId] -> getFinish() == 0)
+ {
+ channels_[channelId] -> handleFinish();
+
+ //
+ // Force a failure in the case somebody
+ // would try to read from the channel.
+ //
+
+ shutdown(getFd(channelId), SHUT_RD);
+
+ //
+ // If the failure was not originated by
+ // the remote, send a channel shutdown
+ // message.
+ //
+
+ if (channels_[channelId] -> getClosing() == 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Finishing channel for FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << " because of failure.\n"
+ << logofs_flush;
+ #endif
+
+ if (handleControl(code_finish_connection, channelId) < 0)
+ {
+ return -1;
+ }
+ }
+ }
+
+ return 1;
+}
+
+int Proxy::handleFinishFromProxy(int channelId)
+{
+ //
+ // Check if this channel has pending
+ // data to send.
+ //
+
+ if (needFlush(channelId) == 1)
+ {
+ #if defined(TEST) || defined(INFO) || defined(FLUSH)
+ *logofs << "Proxy: WARNING! Flushing data for the "
+ << "finishing channel ID#" << channelId
+ << ".\n" << logofs_flush;
+ #endif
+
+ if (handleFrame(frame_data) < 0)
+ {
+ return -1;
+ }
+ }
+
+ //
+ // Mark the channel. We will free its
+ // resources at the next loop and will
+ // send the drop message to the remote.
+ //
+
+ if (channels_[channelId] -> getClosing() == 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Marking channel for FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << " as closing.\n"
+ << logofs_flush;
+ #endif
+
+ channels_[channelId] -> handleClosing();
+ }
+
+ if (channels_[channelId] -> getFinish() == 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Finishing channel for FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << " because of proxy.\n"
+ << logofs_flush;
+ #endif
+
+ channels_[channelId] -> handleFinish();
+ }
+
+ if (handleFinish(channelId) < 0)
+ {
+ return -1;
+ }
+
+ return 1;
+}
+
+int Proxy::handleDropFromProxy(int channelId)
+{
+ //
+ // Only mark the channel.
+ //
+
+ #ifdef TEST
+ *logofs << "Proxy: Marking channel for FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << " as being dropped.\n"
+ << logofs_flush;
+ #endif
+
+ if (channels_[channelId] -> getDrop() == 0)
+ {
+ channels_[channelId] -> handleDrop();
+ }
+
+ return 1;
+}
+
+//
+// Close the channel and deallocate all its
+// resources.
+//
+
+int Proxy::handleDrop(int channelId)
+{
+ //
+ // Check if this channel has pending
+ // data to send.
+ //
+
+ if (needFlush(channelId) == 1)
+ {
+ if (channels_[channelId] -> getFinish() == 1)
+ {
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! The dropping channel ID#"
+ << channelId << " has data to flush.\n"
+ << logofs_flush;
+ #endif
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(FLUSH)
+ *logofs << "Proxy: WARNING! Flushing data for the "
+ << "dropping channel ID#" << channelId
+ << ".\n" << logofs_flush;
+ #endif
+
+ if (handleFrame(frame_data) < 0)
+ {
+ return -1;
+ }
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Dropping channel for FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << ".\n" << logofs_flush;
+ #endif
+
+ if (channels_[channelId] -> getFinish() == 0)
+ {
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! The channel for FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << " was not marked as "
+ << "finishing.\n" << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": The channel for FD#"
+ << getFd(channelId) << " channel ID#"
+ << channelId << " was not marked as "
+ << "finishing.\n";
+
+ channels_[channelId] -> handleFinish();
+ }
+
+ //
+ // Send the channel shutdown message
+ // to the peer proxy.
+ //
+
+ if (channels_[channelId] -> getClosing() == 1)
+ {
+ if (handleControl(code_drop_connection, channelId) < 0)
+ {
+ return -1;
+ }
+ }
+
+ //
+ // Get rid of the channel.
+ //
+
+ if (channels_[channelId] -> getType() != channel_x11)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Closed connection to "
+ << getTypeName(channels_[channelId] -> getType())
+ << " server.\n" << logofs_flush;
+ #endif
+
+ cerr << "Info" << ": Closed connection to "
+ << getTypeName(channels_[channelId] -> getType())
+ << " server.\n";
+ }
+
+ delete channels_[channelId];
+ channels_[channelId] = NULL;
+
+ cleanupChannelMap(channelId);
+
+ //
+ // Get rid of the transport.
+ //
+
+ deallocateTransport(channelId);
+
+ congestions_[channelId] = 0;
+
+ decreaseChannels(channelId);
+
+ //
+ // Check if the channel was the
+ // one currently selected for
+ // output.
+ //
+
+ if (outputChannel_ == channelId)
+ {
+ outputChannel_ = -1;
+ }
+
+ return 1;
+}
+
+//
+// Send an empty message to the remote peer
+// to verify if the link is alive and let
+// the remote proxy detect a congestion.
+//
+
+int Proxy::handlePing()
+{
+ T_timestamp nowTs = getTimestamp();
+
+ #if defined(DEBUG) || defined(PING)
+
+ *logofs << "Proxy: Checking ping at "
+ << strMsTimestamp(nowTs) << logofs_flush;
+
+ *logofs << " with last loop at "
+ << strMsTimestamp(timeouts_.loopTs) << ".\n"
+ << logofs_flush;
+
+ *logofs << "Proxy: Last bytes in at "
+ << strMsTimestamp(timeouts_.readTs) << logofs_flush;
+
+ *logofs << " last bytes out at "
+ << strMsTimestamp(timeouts_.writeTs) << ".\n"
+ << logofs_flush;
+
+ *logofs << "Proxy: Last ping at "
+ << strMsTimestamp(timeouts_.pingTs) << ".\n"
+ << logofs_flush;
+
+ #endif
+
+ //
+ // Be sure we take into account any clock drift. This
+ // can be caused by the user changing the system timer
+ // or by small adjustments introduced by the operating
+ // system making the clock go backward.
+ //
+
+ if (checkDiffTimestamp(timeouts_.loopTs, nowTs) == 0)
+ {
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Detected drift in system "
+ << "timer. Resetting to current time.\n"
+ << logofs_flush;
+ #endif
+
+ timeouts_.pingTs = nowTs;
+ timeouts_.readTs = nowTs;
+ timeouts_.writeTs = nowTs;
+ }
+
+ //
+ // Check timestamp of last read from remote proxy. It can
+ // happen that we stayed in the main loop long enough to
+ // have idle timeout expired, for example if the proxy was
+ // stopped and restarted or because of an extremely high
+ // load of the system. In this case we don't complain if
+ // there is something new to read from the remote.
+ //
+
+ int diffIn = diffTimestamp(timeouts_.readTs, nowTs);
+
+ if (diffIn >= (control -> PingTimeout * 2) -
+ control -> LatencyTimeout)
+ {
+ //
+ // Force a read to detect whether the remote proxy
+ // aborted the connection.
+ //
+
+ int result = handleRead();
+
+ if (result < 0)
+ {
+ #if defined(TEST) || defined(INFO) || defined(PING)
+ *logofs << "Proxy: WARNING! Detected shutdown waiting "
+ << "for the ping after " << diffIn / 1000
+ << " seconds.\n" << logofs_flush;
+ #endif
+
+ return -1;
+ }
+ else if (result > 0)
+ {
+ diffIn = diffTimestamp(timeouts_.readTs, nowTs);
+
+ if (handleFlush() < 0)
+ {
+ return -1;
+ }
+ }
+ }
+
+ if (diffIn >= (control -> PingTimeout * 2) -
+ control -> LatencyTimeout)
+ {
+ #if defined(TEST) || defined(INFO) || defined(PING)
+ *logofs << "Proxy: Detected congestion at "
+ << strMsTimestamp() << " with " << diffIn / 1000
+ << " seconds since the last read.\n"
+ << logofs_flush;
+ #endif
+
+ //
+ // There are two types of proxy congestion. The first,
+ // affecting the ability of the proxy to write the
+ // encoded data to the network, is controlled by the
+ // congestion_ flag. The flag is raised when no data
+ // is received from the remote proxy within a timeout.
+ // On the X client side, the flag is also raised when
+ // the proxy runs out of tokens.
+ //
+
+ if (control -> ProxyMode == proxy_server)
+ {
+ //
+ // At X server side we must return to read data
+ // from the channels after a while, because we
+ // need to give a chance to the channel to read
+ // the key sequence CTRL+ALT+SHIFT+ESC.
+ //
+
+ if (congestion_ == 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Forcibly entering congestion due to "
+ << "timeout with " << tokens_[token_control].remaining
+ << " tokens.\n" << logofs_flush;
+ #endif
+
+ congestion_ = 1;
+ }
+ else
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Forcibly exiting congestion due to "
+ << "timeout with " << tokens_[token_control].remaining
+ << " tokens.\n" << logofs_flush;
+ #endif
+
+ congestion_ = 0;
+ }
+ }
+ else
+ {
+ #if defined(TEST) || defined(INFO)
+
+ if (congestion_ == 0)
+ {
+ *logofs << "Proxy: Entering congestion due to timeout "
+ << "with " << tokens_[token_control].remaining
+ << " tokens.\n" << logofs_flush;
+ }
+
+ #endif
+
+ congestion_ = 1;
+ }
+
+ if (control -> ProxyTimeout > 0 &&
+ diffIn >= (control -> ProxyTimeout -
+ control -> LatencyTimeout))
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! No data received from "
+ << "remote proxy on FD#" << fd_ << " within "
+ << (diffIn + control -> LatencyTimeout) / 1000
+ << " seconds.\n" << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": No data received from remote "
+ << "proxy within " << (diffIn + control ->
+ LatencyTimeout) / 1000 << " seconds.\n";
+
+ HandleAbort();
+ }
+ else
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: WARNING! No data received from "
+ << "remote proxy on FD#" << fd_ << " since "
+ << diffIn << " Ms.\n" << logofs_flush;
+ #endif
+
+ if (control -> ProxyTimeout > 0 &&
+ isTimestamp(timeouts_.alertTs) == 0 &&
+ diffIn >= (control -> ProxyTimeout -
+ control -> LatencyTimeout) / 4)
+ {
+ //
+ // If we are in the middle of a shutdown
+ // procedure but the remote is not resp-
+ // onding, force the closure of the link.
+ //
+
+ if (finish_ != 0)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! No response received from "
+ << "the remote proxy on FD#" << fd_ << " while "
+ << "waiting for the shutdown.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": No response received from remote "
+ << "proxy while waiting for the shutdown.\n";
+
+ HandleAbort();
+ }
+ else
+ {
+ cerr << "Warning" << ": No data received from remote "
+ << "proxy within " << (diffIn + control ->
+ LatencyTimeout) / 1000 << " seconds.\n";
+
+ if (alert_ == 0)
+ {
+ if (control -> ProxyMode == proxy_client)
+ {
+ alert_ = CLOSE_DEAD_PROXY_CONNECTION_CLIENT_ALERT;
+ }
+ else
+ {
+ alert_ = CLOSE_DEAD_PROXY_CONNECTION_SERVER_ALERT;
+ }
+
+ HandleAlert(alert_, 1);
+ }
+
+ timeouts_.alertTs = nowTs;
+ }
+ }
+ }
+ }
+
+ //
+ // Check if we need to update the congestion
+ // counter.
+ //
+
+ int diffOut = diffTimestamp(timeouts_.writeTs, nowTs);
+
+ if (agent_ != nothing && congestions_[agent_] == 0 &&
+ statistics -> getCongestionInFrame() >= 1 &&
+ diffOut >= (control -> IdleTimeout -
+ control -> LatencyTimeout * 5))
+ {
+ #if defined(TEST) || defined(INFO) || defined(PING)
+ *logofs << "Proxy: Forcing an update of the "
+ << "congestion counter after timeout.\n"
+ << logofs_flush;
+ #endif
+
+ statistics -> updateCongestion(tokens_[token_control].remaining,
+ tokens_[token_control].limit);
+ }
+
+ //
+ // Send a new token if we didn't send any data to
+ // the remote for longer than the ping timeout.
+ // The client side sends a token, the server side
+ // responds with a token reply.
+ //
+ // VMWare virtual machines can have the system
+ // timer deadly broken. Try to send a ping regard-
+ // less we are the client or the server proxy to
+ // force a write by the remote.
+ //
+
+ if (control -> ProxyMode == proxy_client ||
+ diffIn >= (control -> PingTimeout * 4) -
+ control -> LatencyTimeout)
+ {
+ //
+ // We need to send a new ping even if we didn't
+ // receive anything from the remote within the
+ // ping timeout. The server side will respond
+ // to our ping, so we use the ping to force the
+ // remote end to send some data.
+ //
+
+ if (diffIn >= (control -> PingTimeout -
+ control -> LatencyTimeout * 5) ||
+ diffOut >= (control -> PingTimeout -
+ control -> LatencyTimeout * 5))
+ {
+ int diffPing = diffTimestamp(timeouts_.pingTs, nowTs);
+
+ if (diffPing >= (control -> PingTimeout -
+ control -> LatencyTimeout * 5))
+ {
+ if (tokens_[token_control].remaining > 0)
+ {
+ #if defined(TEST) || defined(INFO) || defined(PING)
+ *logofs << "Proxy: Sending a new ping at " << strMsTimestamp()
+ << " with " << tokens_[token_control].remaining
+ << " tokens and elapsed in " << diffIn << " out "
+ << diffOut << " ping " << diffPing
+ << ".\n" << logofs_flush;
+ #endif
+
+ if (handleFrame(frame_ping) < 0)
+ {
+ return -1;
+ }
+
+ timeouts_.pingTs = nowTs;
+ }
+ #if defined(TEST) || defined(INFO) || defined(PING)
+ else
+ {
+ *logofs << "Proxy: WARNING! Can't send a new ping "
+ << "with no tokens and elapsed in " << diffIn
+ << " out " << diffOut << " ping " << diffPing
+ << ".\n" << logofs_flush;
+ }
+ #endif
+ }
+ #if defined(TEST) || defined(INFO) || defined(PING)
+ else
+ {
+ *logofs << "Proxy: Not sending a new ping with "
+ << "elapsed in " << diffIn << " out "
+ << diffOut << " ping " << diffPing
+ << ".\n" << logofs_flush;
+ }
+ #endif
+ }
+ }
+
+ return 1;
+}
+
+int Proxy::handleSyncFromProxy(int channelId)
+{
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: WARNING! Received a synchronization "
+ << "request from the remote proxy.\n"
+ << logofs_flush;
+ #endif
+
+ if (handleControl(code_sync_reply, channelId) < 0)
+ {
+ return -1;
+ }
+
+ return 1;
+}
+
+int Proxy::handleResetStores()
+{
+ //
+ // Recreate the message stores.
+ //
+
+ delete clientStore_;
+ delete serverStore_;
+
+ clientStore_ = new ClientStore(compressor_);
+ serverStore_ = new ServerStore(compressor_);
+
+ timeouts_.loadTs = nullTimestamp();
+
+ //
+ // Replace message stores in channels.
+ //
+
+ T_list &channelList = activeChannels_.getList();
+
+ for (T_list::iterator j = channelList.begin();
+ j != channelList.end(); j++)
+ {
+ int channelId = *j;
+
+ if (channels_[channelId] != NULL)
+ {
+ if (channels_[channelId] -> setStores(clientStore_, serverStore_) < 0)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Failed to replace message stores in "
+ << "channel for FD#" << getFd(channelId) << ".\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Failed to replace message stores in "
+ << "channel for FD#" << getFd(channelId) << ".\n";
+
+ return -1;
+ }
+ #ifdef TEST
+ else
+ {
+ *logofs << "Proxy: Replaced message stores in channel "
+ << "for FD#" << getFd(channelId) << ".\n"
+ << logofs_flush;
+ }
+ #endif
+ }
+ }
+
+ return 1;
+}
+
+int Proxy::handleResetPersistentCache()
+{
+ char *fullName = new char[strlen(control -> PersistentCachePath) +
+ strlen(control -> PersistentCacheName) + 2];
+
+ strcpy(fullName, control -> PersistentCachePath);
+ strcat(fullName, "/");
+ strcat(fullName, control -> PersistentCacheName);
+
+ #ifdef TEST
+ *logofs << "Proxy: Going to remove persistent cache file '"
+ << fullName << "'\n" << logofs_flush;
+ #endif
+
+ unlink(fullName);
+
+ delete [] fullName;
+
+ delete [] control -> PersistentCacheName;
+
+ control -> PersistentCacheName = NULL;
+
+ return 1;
+}
+
+void Proxy::handleResetFlush()
+{
+ #ifdef TEST
+ *logofs << "Proxy: Going to reset flush counters "
+ << "for proxy FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ //
+ // Reset the proxy priority flag.
+ //
+
+ priority_ = 0;
+
+ //
+ // Restore buffers to their initial
+ // size.
+ //
+
+ transport_ -> partialReset();
+
+ //
+ // Update the timestamp of the last
+ // write operation performed on the
+ // socket.
+ //
+
+ timeouts_.writeTs = getTimestamp();
+}
+
+int Proxy::handleFinish()
+{
+ //
+ // Reset the timestamps to give the proxy
+ // another chance to show the 'no response'
+ // dialog if the shutdown message doesn't
+ // come in time.
+ //
+
+ timeouts_.readTs = getTimestamp();
+
+ timeouts_.alertTs = nullTimestamp();
+
+ finish_ = 1;
+
+ return 1;
+}
+
+int Proxy::handleShutdown()
+{
+ //
+ // Send shutdown message to remote proxy.
+ //
+
+ shutdown_ = 1;
+
+ handleControl(code_shutdown_request);
+
+ #ifdef TEST
+ *logofs << "Proxy: Starting shutdown procedure "
+ << "for proxy FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ //
+ // Ensure that all the data accumulated
+ // in the transport buffer is flushed
+ // to the network layer.
+ //
+
+ for (int i = 0; i < 100; i++)
+ {
+ if (canFlush() == 1)
+ {
+ handleFlush();
+ }
+ else
+ {
+ break;
+ }
+
+ usleep(100000);
+ }
+
+ //
+ // Now wait for the network layers to
+ // consume all the data.
+ //
+
+ for (int i = 0; i < 100; i++)
+ {
+ if (transport_ -> queued() <= 0)
+ {
+ break;
+ }
+
+ usleep(100000);
+ }
+
+ //
+ // Give time to the remote end to read
+ // the shutdown message and close the
+ // connection.
+ //
+
+ transport_ -> wait(10000);
+
+ #ifdef TEST
+ *logofs << "Proxy: Ending shutdown procedure "
+ << "for proxy FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ return 1;
+}
+
+int Proxy::handleChannelConfiguration()
+{
+ if (activeChannels_.getSize() == 0)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Going to initialize the static "
+ << "members in channels for proxy FD#"
+ << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ Channel::setReferences();
+
+ ClientChannel::setReferences();
+ ServerChannel::setReferences();
+
+ GenericChannel::setReferences();
+ }
+
+ return 1;
+}
+
+int Proxy::handleSocketConfiguration()
+{
+ //
+ // Set linger mode on proxy to correctly
+ // get shutdown notification.
+ //
+
+ SetLingerTimeout(fd_, 30);
+
+ //
+ // Set keep-alive on socket so that if remote link
+ // terminates abnormally (as killed hard or because
+ // of a power-off) process will get a SIGPIPE. In
+ // practice this is useless as proxies already ping
+ // each other every few seconds.
+ //
+
+ if (control -> OptionProxyKeepAlive == 1)
+ {
+ SetKeepAlive(fd_);
+ }
+
+ //
+ // Set 'priority' flag at TCP layer for path
+ // proxy-to-proxy. Look at IPTOS_LOWDELAY in
+ // man 7 ip.
+ //
+
+ if (control -> OptionProxyLowDelay == 1)
+ {
+ SetLowDelay(fd_);
+ }
+
+ //
+ // Update size of TCP send and receive buffers.
+ //
+
+ if (control -> OptionProxySendBuffer != -1)
+ {
+ SetSendBuffer(fd_, control -> OptionProxySendBuffer);
+ }
+
+ if (control -> OptionProxyReceiveBuffer != -1)
+ {
+ SetReceiveBuffer(fd_, control -> OptionProxyReceiveBuffer);
+ }
+
+ //
+ // Update TCP_NODELAY settings. Note that on old Linux
+ // kernels turning off the Nagle algorithm didn't work
+ // when proxy was run through a PPP link. Trying to do
+ // so caused the kernel to stop delivering data to us
+ // if a serious network congestion was encountered.
+ //
+
+ if (control -> ProxyMode == proxy_client)
+ {
+ if (control -> OptionProxyClientNoDelay != -1)
+ {
+ SetNoDelay(fd_, control -> OptionProxyClientNoDelay);
+ }
+ }
+ else
+ {
+ if (control -> OptionProxyServerNoDelay != -1)
+ {
+ SetNoDelay(fd_, control -> OptionProxyServerNoDelay);
+ }
+ }
+
+ return 1;
+}
+
+int Proxy::handleLinkConfiguration()
+{
+ #ifdef TEST
+ *logofs << "Proxy: Propagating parameters to "
+ << "channels' read buffers.\n"
+ << logofs_flush;
+ #endif
+
+ T_list &channelList = activeChannels_.getList();
+
+ for (T_list::iterator j = channelList.begin();
+ j != channelList.end(); j++)
+ {
+ int channelId = *j;
+
+ if (channels_[channelId] != NULL)
+ {
+ channels_[channelId] -> handleConfiguration();
+ }
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Propagating parameters to "
+ << "proxy buffers.\n"
+ << logofs_flush;
+ #endif
+
+ readBuffer_.setSize(control -> ProxyInitialReadSize,
+ control -> ProxyMaximumBufferSize);
+
+ encodeBuffer_.setSize(control -> TransportProxyBufferSize,
+ control -> TransportProxyBufferThreshold,
+ control -> TransportMaximumBufferSize);
+
+ transport_ -> setSize(control -> TransportProxyBufferSize,
+ control -> TransportProxyBufferThreshold,
+ control -> TransportMaximumBufferSize);
+
+ #ifdef TEST
+ *logofs << "Proxy: Configuring the proxy timeouts.\n"
+ << logofs_flush;
+ #endif
+
+ timeouts_.split = control -> SplitTimeout;
+ timeouts_.motion = control -> MotionTimeout;
+
+ #ifdef TEST
+ *logofs << "Proxy: Configuring the proxy tokens.\n"
+ << logofs_flush;
+ #endif
+
+ tokens_[token_control].size = control -> TokenSize;
+ tokens_[token_control].limit = control -> TokenLimit;
+
+ if (tokens_[token_control].limit < 1)
+ {
+ tokens_[token_control].limit = 1;
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(LIMIT)
+ *logofs << "Proxy: TOKEN! LIMIT! Setting token ["
+ << DumpToken(token_control) << "] size to "
+ << tokens_[token_control].size << " and limit to "
+ << tokens_[token_control].limit << ".\n"
+ << logofs_flush;
+ #endif
+
+ tokens_[token_split].size = control -> TokenSize;
+ tokens_[token_split].limit = control -> TokenLimit / 2;
+
+ if (tokens_[token_split].limit < 1)
+ {
+ tokens_[token_split].limit = 1;
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(LIMIT)
+ *logofs << "Proxy: TOKEN! LIMIT! Setting token ["
+ << DumpToken(token_split) << "] size to "
+ << tokens_[token_split].size << " and limit to "
+ << tokens_[token_split].limit << ".\n"
+ << logofs_flush;
+ #endif
+
+ tokens_[token_data].size = control -> TokenSize;
+ tokens_[token_data].limit = control -> TokenLimit / 4;
+
+ if (tokens_[token_data].limit < 1)
+ {
+ tokens_[token_data].limit = 1;
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(LIMIT)
+ *logofs << "Proxy: TOKEN! LIMIT! Setting token ["
+ << DumpToken(token_data) << "] size to "
+ << tokens_[token_data].size << " and limit to "
+ << tokens_[token_data].limit << ".\n"
+ << logofs_flush;
+ #endif
+
+ for (int i = token_control; i <= token_data; i++)
+ {
+ tokens_[i].remaining = tokens_[i].limit;
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(LIMIT)
+ *logofs << "Proxy: LIMIT! Using client bitrate "
+ << "limit " << control -> ClientBitrateLimit
+ << " server bitrate limit " << control ->
+ ServerBitrateLimit << " with local limit "
+ << control -> LocalBitrateLimit << ".\n"
+ << logofs_flush;
+ #endif
+
+ //
+ // Set the other parameters based on
+ // the token size.
+ //
+
+ int base = control -> TokenSize;
+
+ control -> SplitDataThreshold = base * 4;
+ control -> SplitDataPacketLimit = base / 2;
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: LIMIT! Setting split data threshold "
+ << "to " << control -> SplitDataThreshold
+ << " split packet limit to " << control ->
+ SplitDataPacketLimit << " with base "
+ << base << ".\n" << logofs_flush;
+ #endif
+
+ //
+ // Set the number of bytes read from the
+ // data channels at each loop. This will
+ // basically determine the maximum band-
+ // width available for the generic chan-
+ // nels.
+ //
+
+ control -> GenericInitialReadSize = base / 2;
+ control -> GenericMaximumBufferSize = base / 2;
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: LIMIT! Setting generic channel "
+ << "initial read size to " << control ->
+ GenericInitialReadSize << " maximum read "
+ << "size to " << control -> GenericMaximumBufferSize
+ << " with base " << base << ".\n"
+ << logofs_flush;
+ #endif
+
+ return 1;
+}
+
+int Proxy::handleCacheConfiguration()
+{
+ #ifdef TEST
+ *logofs << "Proxy: Configuring cache according to pack parameters.\n"
+ << logofs_flush;
+ #endif
+
+ //
+ // Further adjust the cache parameters. If
+ // packing of the images is enabled, reduce
+ // the size available for plain images.
+ //
+
+ if (control -> SessionMode == session_agent)
+ {
+ if (control -> PackMethod != NO_PACK)
+ {
+ clientStore_ -> getRequestStore(X_PutImage) ->
+ cacheThreshold = PUTIMAGE_CACHE_THRESHOLD_IF_PACKED;
+
+ clientStore_ -> getRequestStore(X_PutImage) ->
+ cacheLowerThreshold = PUTIMAGE_CACHE_LOWER_THRESHOLD_IF_PACKED;
+ }
+ }
+
+ //
+ // If this is a shadow session increase the
+ // size of the image cache.
+ //
+
+ if (control -> SessionMode == session_shadow)
+ {
+ if (control -> PackMethod != NO_PACK)
+ {
+ clientStore_ -> getRequestStore(X_NXPutPackedImage) ->
+ cacheThreshold = PUTPACKEDIMAGE_CACHE_THRESHOLD_IF_PACKED_SHADOW;
+
+ clientStore_ -> getRequestStore(X_NXPutPackedImage) ->
+ cacheLowerThreshold = PUTPACKEDIMAGE_CACHE_LOWER_THRESHOLD_IF_PACKED_SHADOW;
+ }
+ else
+ {
+ clientStore_ -> getRequestStore(X_PutImage) ->
+ cacheThreshold = PUTIMAGE_CACHE_THRESHOLD_IF_SHADOW;
+
+ clientStore_ -> getRequestStore(X_PutImage) ->
+ cacheLowerThreshold = PUTIMAGE_CACHE_LOWER_THRESHOLD_IF_SHADOW;
+ }
+ }
+
+ return 1;
+}
+
+int Proxy::handleSaveStores()
+{
+ //
+ // Save content of stores on disk.
+ //
+
+ char *cacheToAdopt = NULL;
+
+ if (control -> PersistentCacheEnableSave)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Going to save content of client store.\n"
+ << logofs_flush;
+ #endif
+
+ cacheToAdopt = handleSaveAllStores(control -> PersistentCachePath);
+ }
+ #ifdef TEST
+ else
+ {
+ if (control -> ProxyMode == proxy_client)
+ {
+ *logofs << "Proxy: Saving persistent cache to disk disabled.\n"
+ << logofs_flush;
+ }
+ else
+ {
+ *logofs << "Proxy: PANIC! Protocol violation in command save.\n"
+ << logofs_flush;
+
+ cerr << "Error" << ": Protocol violation in command save.\n";
+
+ HandleCleanup();
+ }
+ }
+ #endif
+
+ if (cacheToAdopt != NULL)
+ {
+ //
+ // Do we have a cache already?
+ //
+
+ if (control -> PersistentCacheName != NULL)
+ {
+ //
+ // Check if old and new cache are the same.
+ // In this case don't remove the old cache.
+ //
+
+ if (strcasecmp(control -> PersistentCacheName, cacheToAdopt) != 0)
+ {
+ handleResetPersistentCache();
+ }
+
+ delete [] control -> PersistentCacheName;
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Setting current persistent cache file to '"
+ << cacheToAdopt << "'\n" << logofs_flush;
+ #endif
+
+ control -> PersistentCacheName = cacheToAdopt;
+
+ return 1;
+ }
+ #ifdef TEST
+ else
+ {
+ *logofs << "Proxy: No cache file produced from message stores.\n"
+ << logofs_flush;
+ }
+ #endif
+
+ //
+ // It can be that we didn't generate a new cache
+ // because store was too small or persistent cache
+ // was disabled. This is not an error.
+ //
+
+ return 0;
+}
+
+int Proxy::handleLoadStores()
+{
+ //
+ // Restore the content of the client store
+ // from disk if a valid cache was negotiated
+ // at session startup.
+ //
+
+ if (control -> PersistentCacheEnableLoad == 1 &&
+ control -> PersistentCachePath != NULL &&
+ control -> PersistentCacheName != NULL)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Going to load content of client store.\n"
+ << logofs_flush;
+ #endif
+
+ //
+ // Returns the same string passed as name of
+ // the cache, or NULL if it was not possible
+ // to load the cache from disk.
+ //
+
+ if (handleLoadAllStores(control -> PersistentCachePath,
+ control -> PersistentCacheName) == NULL)
+ {
+ //
+ // The corrupted cache should have been
+ // removed from disk. Get rid of the
+ // reference so we don't try to delete
+ // it once again.
+ //
+
+ if (control -> PersistentCacheName != NULL)
+ {
+ delete [] control -> PersistentCacheName;
+ }
+
+ control -> PersistentCacheName = NULL;
+
+ return -1;
+ }
+
+ //
+ // Set timestamp of last time cache
+ // was loaded from data on disk.
+ //
+
+ timeouts_.loadTs = getTimestamp();
+
+ return 1;
+ }
+ #ifdef TEST
+ else
+ {
+ if (control -> ProxyMode == proxy_client)
+ {
+ *logofs << "Proxy: Loading of cache disabled or no cache file selected.\n"
+ << logofs_flush;
+ }
+ else
+ {
+ *logofs << "Proxy: PANIC! Protocol violation in command load.\n"
+ << logofs_flush;
+
+ cerr << "Error" << ": Protocol violation in command load.\n";
+
+ HandleCleanup();
+ }
+ }
+ #endif
+
+ return 0;
+}
+
+int Proxy::handleControl(T_proxy_code code, int data)
+{
+ //
+ // Send the given control messages
+ // to the remote proxy.
+ //
+
+ #if defined(TEST) || defined(INFO)
+
+ if (data != -1)
+ {
+ if (code == code_control_token_reply ||
+ code == code_split_token_reply ||
+ code == code_data_token_reply)
+ {
+ *logofs << "Proxy: TOKEN! Sending message '" << DumpControl(code)
+ << "' at " << strMsTimestamp() << " with count "
+ << data << ".\n" << logofs_flush;
+ }
+ else
+ {
+ *logofs << "Proxy: Sending message '" << DumpControl(code)
+ << "' at " << strMsTimestamp() << " with data ID#"
+ << data << ".\n" << logofs_flush;
+ }
+ }
+ else
+ {
+ *logofs << "Proxy: Sending message '" << DumpControl(code)
+ << "' at " << strMsTimestamp() << ".\n"
+ << logofs_flush;
+ }
+
+ #endif
+
+ //
+ // Add the control message and see if the
+ // data has to be flushed immediately.
+ //
+
+ if (addControlCodes(code, data) < 0)
+ {
+ return -1;
+ }
+
+ switch (code)
+ {
+ //
+ // Append the first data read from the opened
+ // channel to the control code.
+ //
+
+ case code_new_x_connection:
+ case code_new_cups_connection:
+ case code_new_aux_connection:
+ case code_new_smb_connection:
+ case code_new_media_connection:
+ case code_new_http_connection:
+ case code_new_font_connection:
+ case code_new_slave_connection:
+
+ //
+ // Do we send the token reply immediately?
+ // The control messages are put at the begin-
+ // ning of the of the encode buffer, so we may
+ // reply to multiple tokens before having the
+ // chance of handling the actual frame data.
+ // On the other hand, the sooner we reply, the
+ // sooner the remote proxy is restarted.
+ //
+
+ case code_control_token_reply:
+ case code_split_token_reply:
+ case code_data_token_reply:
+ {
+ break;
+ }
+
+ //
+ // Also send the congestion control codes
+ // immediately.
+ //
+ // case code_begin_congestion:
+ // case code_end_congestion:
+ //
+
+ default:
+ {
+ priority_ = 1;
+
+ break;
+ }
+ }
+
+ if (priority_ == 1)
+ {
+ if (handleFrame(frame_data) < 0)
+ {
+ return -1;
+ }
+ }
+
+ return 1;
+}
+
+int Proxy::handleSwitch(int channelId)
+{
+ //
+ // If data is for a different channel than last
+ // selected for output, prepend to the data the
+ // new channel id.
+ //
+
+ #ifdef DEBUG
+ *logofs << "Proxy: Requested a switch with "
+ << "current channel ID#" << outputChannel_
+ << " new channel ID#" << channelId << ".\n"
+ << logofs_flush;
+ #endif
+
+ if (channelId != outputChannel_)
+ {
+ if (needFlush() == 1)
+ {
+ #if defined(TEST) || defined(INFO) || defined(FLUSH)
+ *logofs << "Proxy: WARNING! Flushing data for the "
+ << "previous channel ID#" << outputChannel_
+ << ".\n" << logofs_flush;
+ #endif
+
+ if (handleFrame(frame_data) < 0)
+ {
+ return -1;
+ }
+ }
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Sending message '"
+ << DumpControl(code_switch_connection) << "' at "
+ << strMsTimestamp() << " with FD#" << getFd(channelId)
+ << " channel ID#" << channelId << ".\n"
+ << logofs_flush;
+ #endif
+
+ if (addControlCodes(code_switch_connection, channelId) < 0)
+ {
+ return -1;
+ }
+
+ outputChannel_ = channelId;
+ }
+
+ return 1;
+}
+
+int Proxy::addTokenCodes(T_proxy_token &token)
+{
+ #if defined(TEST) || defined(INFO) || defined(TOKEN)
+ *logofs << "Proxy: TOKEN! Sending token ["
+ << DumpToken(token.type) << "] with "
+ << token.bytes << " bytes accumulated size "
+ << token.size << " and " << token.remaining
+ << " available.\n" << logofs_flush;
+ #endif
+
+ //
+ // Give a 'weight' to the token. The tokens
+ // remaining can become negative if we sent
+ // a packet that was exceptionally big.
+ //
+
+ int count = 0;
+
+ if (control -> isProtoStep7() == 1)
+ {
+ count = token.bytes / token.size;
+
+ if (count > 255)
+ {
+ count = 255;
+ }
+ }
+
+ //
+ // Force a count of 1, for example
+ // if this is a ping.
+ //
+
+ if (count < 1)
+ {
+ count = 1;
+
+ token.bytes = 0;
+ }
+ else
+ {
+ //
+ // Let the next token account for the
+ // remaining bytes.
+ //
+
+ token.bytes %= token.size;
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(TOKEN)
+ *logofs << "Proxy: Sending message '"
+ << DumpControl(token.request) << "' at "
+ << strMsTimestamp() << " with count " << count
+ << ".\n" << logofs_flush;
+ #endif
+
+ controlCodes_[controlLength_++] = 0;
+ controlCodes_[controlLength_++] = (unsigned char) token.request;
+ controlCodes_[controlLength_++] = (unsigned char) count;
+
+ statistics -> addFrameOut();
+
+ token.remaining -= count;
+
+ return 1;
+}
+
+int Proxy::handleToken(T_frame_type type)
+{
+ #if defined(TEST) || defined(INFO) || defined(TOKEN)
+ *logofs << "Proxy: TOKEN! Checking tokens with "
+ << "frame type [";
+
+ *logofs << (type == frame_ping ? "frame_ping" : "frame_data");
+
+ *logofs << "] with stream ratio " << statistics ->
+ getStreamRatio() << ".\n" << logofs_flush;
+ #endif
+
+ if (type == frame_data)
+ {
+ if (control -> isProtoStep7() == 1)
+ {
+ //
+ // Send a distinct token for each data type.
+ // We don't want to slow down the sending of
+ // the X events, X replies and split confir-
+ // mation events on the X server side, so
+ // take care only of the generic data token.
+ //
+
+ if (control -> ProxyMode == proxy_client)
+ {
+ statistics -> updateControlToken(tokens_[token_control].bytes);
+
+ if (tokens_[token_control].bytes > tokens_[token_control].size)
+ {
+ if (addTokenCodes(tokens_[token_control]) < 0)
+ {
+ return -1;
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(TOKEN)
+
+ T_proxy_token &token = tokens_[token_control];
+
+ *logofs << "Proxy: TOKEN! Token class ["
+ << DumpToken(token.type) << "] has now "
+ << token.bytes << " bytes accumulated and "
+ << token.remaining << " tokens remaining.\n"
+ << logofs_flush;
+ #endif
+ }
+
+ statistics -> updateSplitToken(tokens_[token_split].bytes);
+
+ if (tokens_[token_split].bytes > tokens_[token_split].size)
+ {
+ if (addTokenCodes(tokens_[token_split]) < 0)
+ {
+ return -1;
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(TOKEN)
+
+ T_proxy_token &token = tokens_[token_split];
+
+ *logofs << "Proxy: TOKEN! Token class ["
+ << DumpToken(token.type) << "] has now "
+ << token.bytes << " bytes accumulated and "
+ << token.remaining << " tokens remaining.\n"
+ << logofs_flush;
+ #endif
+ }
+ }
+
+ statistics -> updateDataToken(tokens_[token_data].bytes);
+
+ if (tokens_[token_data].bytes > tokens_[token_data].size)
+ {
+ if (addTokenCodes(tokens_[token_data]) < 0)
+ {
+ return -1;
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(TOKEN)
+
+ T_proxy_token &token = tokens_[token_data];
+
+ *logofs << "Proxy: TOKEN! Token class ["
+ << DumpToken(token.type) << "] has now "
+ << token.bytes << " bytes accumulated and "
+ << token.remaining << " tokens remaining.\n"
+ << logofs_flush;
+ #endif
+ }
+ }
+ else
+ {
+ //
+ // Sum everything to the control token.
+ //
+
+ if (control -> ProxyMode == proxy_client)
+ {
+ statistics -> updateControlToken(tokens_[token_control].bytes);
+ statistics -> updateSplitToken(tokens_[token_control].bytes);
+ statistics -> updateDataToken(tokens_[token_control].bytes);
+
+ if (tokens_[token_control].bytes > tokens_[token_control].size)
+ {
+ if (addTokenCodes(tokens_[token_control]) < 0)
+ {
+ return -1;
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(TOKEN)
+
+ T_proxy_token &token = tokens_[token_control];
+
+ *logofs << "Proxy: TOKEN! Token class ["
+ << DumpToken(token.type) << "] has now "
+ << token.bytes << " bytes accumulated and "
+ << token.remaining << " tokens remaining.\n"
+ << logofs_flush;
+ #endif
+ }
+ }
+ }
+ }
+ else
+ {
+ if (addTokenCodes(tokens_[token_control]) < 0)
+ {
+ return -1;
+ }
+
+ //
+ // Reset all counters on a ping.
+ //
+
+ tokens_[token_control].bytes = 0;
+ tokens_[token_split].bytes = 0;
+ tokens_[token_data].bytes = 0;
+
+ #if defined(TEST) || defined(INFO) || defined(TOKEN)
+
+ T_proxy_token &token = tokens_[token_control];
+
+ *logofs << "Proxy: TOKEN! Token class ["
+ << DumpToken(token.type) << "] has now "
+ << token.bytes << " bytes accumulated and "
+ << token.remaining << " tokens remaining.\n"
+ << logofs_flush;
+ #endif
+ }
+
+ //
+ // Check if we have entered in
+ // congestion state.
+ //
+
+ if (congestion_ == 0 &&
+ tokens_[token_control].remaining <= 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Entering congestion with "
+ << tokens_[token_control].remaining
+ << " tokens remaining.\n" << logofs_flush;
+ #endif
+
+ congestion_ = 1;
+ }
+
+ statistics -> updateCongestion(tokens_[token_control].remaining,
+ tokens_[token_control].limit);
+
+ return 1;
+}
+
+int Proxy::handleTokenFromProxy(T_proxy_token &token, int count)
+{
+ #if defined(TEST) || defined(INFO) || defined(TOKEN)
+ *logofs << "Proxy: TOKEN! Received token ["
+ << DumpToken(token.type) << "] request at "
+ << strMsTimestamp() << " with count "
+ << count << ".\n" << logofs_flush;
+ #endif
+
+ if (control -> isProtoStep7() == 0)
+ {
+ if (control -> ProxyMode == proxy_client ||
+ token.request != code_control_token_request)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Invalid token request received from remote.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Invalid token request received from remote.\n";
+
+ HandleCleanup();
+ }
+ }
+
+ //
+ // Add our token reply.
+ //
+
+ if (handleControl(token.reply, count) < 0)
+ {
+ return -1;
+ }
+
+ return 1;
+}
+
+int Proxy::handleTokenReplyFromProxy(T_proxy_token &token, int count)
+{
+ #if defined(TEST) || defined(INFO) || defined(TOKEN)
+ *logofs << "Proxy: TOKEN! Received token ["
+ << DumpToken(token.type) << "] reply at "
+ << strMsTimestamp() << " with count " << count
+ << ".\n" << logofs_flush;
+ #endif
+
+ //
+ // Increment the available tokens.
+ //
+
+ if (control -> isProtoStep7() == 0)
+ {
+ if (token.reply != code_control_token_reply)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Invalid token reply received from remote.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Invalid token reply received from remote.\n";
+
+ HandleCleanup();
+ }
+
+ count = 1;
+ }
+
+ token.remaining += count;
+
+ if (token.remaining > token.limit)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Token overflow handling messages.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Token overflow handling messages.\n";
+
+ HandleCleanup();
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(TOKEN)
+ *logofs << "Proxy: TOKEN! Token class ["
+ << DumpToken(token.type) << "] has now " << token.bytes
+ << " bytes accumulated and " << token.remaining
+ << " tokens remaining.\n" << logofs_flush;
+ #endif
+
+ //
+ // Check if we can jump out of the
+ // congestion state.
+ //
+
+ if (congestion_ == 1 &&
+ tokens_[token_control].remaining > 0)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: Exiting congestion with "
+ << tokens_[token_control].remaining
+ << " tokens remaining.\n" << logofs_flush;
+ #endif
+
+ congestion_ = 0;
+ }
+
+ statistics -> updateCongestion(tokens_[token_control].remaining,
+ tokens_[token_control].limit);
+
+ return 1;
+}
+
+void Proxy::handleFailOnSave(const char *fullName, const char *failContext) const
+{
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Error saving stores to cache file "
+ << "in context [" << failContext << "].\n" << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Error saving stores to cache file "
+ << "in context [" << failContext << "].\n";
+
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Removing invalid cache '"
+ << fullName << "'.\n" << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Removing invalid cache '"
+ << fullName << "'.\n";
+
+ unlink(fullName);
+}
+
+void Proxy::handleFailOnLoad(const char *fullName, const char *failContext) const
+{
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Error loading stores from cache file "
+ << "in context [" << failContext << "].\n" << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Error loading stores from cache file "
+ << "in context [" << failContext << "].\n";
+
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Removing invalid cache '"
+ << fullName << "'.\n" << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Removing invalid cache '"
+ << fullName << "'.\n";
+
+ unlink(fullName);
+}
+
+int Proxy::handleSaveVersion(unsigned char *buffer, int &major,
+ int &minor, int &patch) const
+{
+ if (control -> isProtoStep8() == 1)
+ {
+ major = 3;
+ minor = 0;
+ patch = 0;
+ }
+ else if (control -> isProtoStep7() == 1)
+ {
+ major = 2;
+ minor = 0;
+ patch = 0;
+ }
+ else
+ {
+ major = 1;
+ minor = 4;
+ patch = 0;
+ }
+
+ *(buffer + 0) = major;
+ *(buffer + 1) = minor;
+
+ PutUINT(patch, buffer + 2, storeBigEndian());
+
+ return 1;
+}
+
+int Proxy::handleLoadVersion(const unsigned char *buffer, int &major,
+ int &minor, int &patch) const
+{
+ major = *(buffer + 0);
+ minor = *(buffer + 1);
+
+ patch = GetUINT(buffer + 2, storeBigEndian());
+
+ //
+ // Force the proxy to discard the
+ // incompatible caches.
+ //
+
+ if (control -> isProtoStep8() == 1)
+ {
+ if (major < 3)
+ {
+ return -1;
+ }
+ }
+ else if (control -> isProtoStep7() == 1)
+ {
+ if (major < 2)
+ {
+ return -1;
+ }
+ }
+ else
+ {
+ if (major != 1 && minor != 4)
+ {
+ return -1;
+ }
+ }
+
+ return 1;
+}
+
+char *Proxy::handleSaveAllStores(const char *savePath) const
+{
+ int cumulativeSize = MessageStore::getCumulativeTotalStorageSize();
+
+ if (cumulativeSize < control -> PersistentCacheThreshold)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: Cache not saved as size is "
+ << cumulativeSize << " with threshold set to "
+ << control -> PersistentCacheThreshold
+ << ".\n" << logofs_flush;
+ #endif
+
+ return NULL;
+ }
+ else if (savePath == NULL)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! No name provided for save path.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": No name provided for save path.\n";
+
+ return NULL;
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Going to save content of message stores.\n"
+ << logofs_flush;
+ #endif
+
+ //
+ // Our parent process is likely going to terminate.
+ // Until we finish saving cache we must ignore its
+ // SIGIPE.
+ //
+
+ DisableSignals();
+
+ ofstream *cachefs = NULL;
+
+ md5_state_t *md5StateStream = NULL;
+ md5_byte_t *md5DigestStream = NULL;
+
+ md5_state_t *md5StateClient = NULL;
+ md5_byte_t *md5DigestClient = NULL;
+
+ char *tempName = NULL;
+
+ char md5String[MD5_LENGTH * 2 + 2];
+
+ char fullName[strlen(savePath) + MD5_LENGTH * 2 + 4];
+
+ if (control -> ProxyMode == proxy_client)
+ {
+ tempName = tempnam(savePath, "Z-C-");
+ }
+ else
+ {
+ tempName = tempnam(savePath, "Z-S-");
+ }
+
+ //
+ // Change the mask to make the file only
+ // readable by the user, then restore the
+ // old mask.
+ //
+
+ mode_t fileMode = umask(0077);
+
+ cachefs = new ofstream(tempName, ios::out | ios::binary);
+
+ umask(fileMode);
+
+ if (tempName == NULL || cachefs == NULL)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Can't create temporary file in '"
+ << savePath << "'.\n" << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Can't create temporary file in '"
+ << savePath << "'.\n";
+
+ if (tempName != NULL)
+ {
+ free(tempName);
+ }
+
+ if (cachefs != NULL)
+ {
+ delete cachefs;
+ }
+
+ EnableSignals();
+
+ return NULL;
+ }
+
+ md5StateStream = new md5_state_t();
+ md5DigestStream = new md5_byte_t[MD5_LENGTH];
+
+ md5_init(md5StateStream);
+
+ //
+ // First write the proxy version.
+ //
+
+ unsigned char version[4];
+
+ int major;
+ int minor;
+ int patch;
+
+ handleSaveVersion(version, major, minor, patch);
+
+ #ifdef TEST
+ *logofs << "Proxy: Saving cache using version '"
+ << major << "." << minor << "." << patch
+ << "'.\n" << logofs_flush;
+ #endif
+
+ if (PutData(cachefs, version, 4) < 0)
+ {
+ handleFailOnSave(tempName, "A");
+
+ delete cachefs;
+
+ delete md5StateStream;
+ delete [] md5DigestStream;
+
+ free(tempName);
+
+ EnableSignals();
+
+ return NULL;
+ }
+
+ //
+ // Make space for the calculated MD5 so we
+ // can later rewind the file and write it
+ // at this position.
+ //
+
+ if (PutData(cachefs, md5DigestStream, MD5_LENGTH) < 0)
+ {
+ handleFailOnSave(tempName, "B");
+
+ delete cachefs;
+
+ delete md5StateStream;
+ delete [] md5DigestStream;
+
+ free(tempName);
+
+ EnableSignals();
+
+ return NULL;
+ }
+
+ md5StateClient = new md5_state_t();
+ md5DigestClient = new md5_byte_t[MD5_LENGTH];
+
+ md5_init(md5StateClient);
+
+ #ifdef DUMP
+
+ ofstream *cacheDump = NULL;
+
+ ofstream *tempfs = (ofstream*) logofs;
+
+ char cacheDumpName[DEFAULT_STRING_LENGTH];
+
+ if (control -> ProxyMode == proxy_client)
+ {
+ snprintf(cacheDumpName, DEFAULT_STRING_LENGTH - 1,
+ "%s/client-cache-dump", control -> TempPath);
+ }
+ else
+ {
+ snprintf(cacheDumpName, DEFAULT_STRING_LENGTH - 1,
+ "%s/server-cache-dump", control -> TempPath);
+ }
+
+ *(cacheDumpName + DEFAULT_STRING_LENGTH - 1) = '\0';
+
+ mode_t fileMode = umask(0077);
+
+ cacheDump = new ofstream(cacheDumpName, ios::out);
+
+ umask(fileMode);
+
+ logofs = cacheDump;
+
+ #endif
+
+ //
+ // Use the virtual method of the concrete proxy class.
+ //
+
+ int allSaved = handleSaveAllStores(cachefs, md5StateStream, md5StateClient);
+
+ #ifdef DUMP
+
+ logofs = tempfs;
+
+ delete cacheDump;
+
+ #endif
+
+ if (allSaved == 0)
+ {
+ handleFailOnSave(tempName, "C");
+
+ delete cachefs;
+
+ delete md5StateStream;
+ delete [] md5DigestStream;
+
+ delete md5StateClient;
+ delete [] md5DigestClient;
+
+ free(tempName);
+
+ EnableSignals();
+
+ return NULL;
+ }
+
+ md5_finish(md5StateClient, md5DigestClient);
+
+ for (unsigned int i = 0; i < MD5_LENGTH; i++)
+ {
+ sprintf(md5String + (i * 2), "%02X", md5DigestClient[i]);
+ }
+
+ strcpy(fullName, (control -> ProxyMode == proxy_client) ? "C-" : "S-");
+
+ strcat(fullName, md5String);
+
+ md5_append(md5StateStream, (const md5_byte_t *) fullName, strlen(fullName));
+ md5_finish(md5StateStream, md5DigestStream);
+
+ //
+ // Go to the beginning of file plus
+ // the integer where we wrote our
+ // proxy version.
+ //
+
+ cachefs -> seekp(4);
+
+ if (PutData(cachefs, md5DigestStream, MD5_LENGTH) < 0)
+ {
+ handleFailOnSave(tempName, "D");
+
+ delete cachefs;
+
+ delete md5StateStream;
+ delete [] md5DigestStream;
+
+ delete md5StateClient;
+ delete [] md5DigestClient;
+
+ free(tempName);
+
+ EnableSignals();
+
+ return NULL;
+ }
+
+ delete cachefs;
+
+ //
+ // Copy the resulting cache name without
+ // the path so that we can return it to
+ // the caller.
+ //
+
+ char *cacheName = new char[MD5_LENGTH * 2 + 4];
+
+ strcpy(cacheName, fullName);
+
+ //
+ // Add the path to the full name and move
+ // the cache into the path.
+ //
+
+ strcpy(fullName, savePath);
+ strcat(fullName, (control -> ProxyMode == proxy_client) ? "/C-" : "/S-");
+ strcat(fullName, md5String);
+
+ #ifdef TEST
+ *logofs << "Proxy: Renaming cache file from '"
+ << tempName << "' to '" << fullName
+ << "'.\n" << logofs_flush;
+ #endif
+
+ rename(tempName, fullName);
+
+ delete md5StateStream;
+ delete [] md5DigestStream;
+
+ delete md5StateClient;
+ delete [] md5DigestClient;
+
+ free(tempName);
+
+ //
+ // Restore the original handlers.
+ //
+
+ EnableSignals();
+
+ #ifdef TEST
+ *logofs << "Proxy: Successfully saved cache file '"
+ << cacheName << "'.\n" << logofs_flush;
+ #endif
+
+ //
+ // This must be enabled only for test
+ // because it requires that client and
+ // server reside on the same machine.
+ //
+
+ if (control -> PersistentCacheCheckOnShutdown == 1 &&
+ control -> ProxyMode == proxy_server)
+ {
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: MATCH! Checking if the file '"
+ << fullName << "' matches a client cache.\n"
+ << logofs_flush;
+ #endif
+
+ strcpy(fullName, savePath);
+ strcat(fullName, "/C-");
+ strcat(fullName, md5String);
+
+ struct stat fileStat;
+
+ if (stat(fullName, &fileStat) != 0)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Can't find a client cache "
+ << "with name '" << fullName << "'.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Can't find a client cache "
+ << "with name '" << fullName << "'.\n";
+
+ HandleShutdown();
+ }
+
+ #if defined(TEST) || defined(INFO)
+ *logofs << "Proxy: MATCH! Client cache '" << fullName
+ << "' matches the local cache.\n"
+ << logofs_flush;
+ #endif
+ }
+
+ return cacheName;
+}
+
+const char *Proxy::handleLoadAllStores(const char *loadPath, const char *loadName) const
+{
+ #ifdef TEST
+ *logofs << "Proxy: Going to load content of message stores.\n"
+ << logofs_flush;
+ #endif
+
+ //
+ // Until we finish loading cache we
+ // must at least ignore any SIGIPE.
+ //
+
+ DisableSignals();
+
+ if (loadPath == NULL || loadName == NULL)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! No path or no file name provided for cache to restore.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": No path or no file name provided for cache to restore.\n";
+
+ EnableSignals();
+
+ return NULL;
+ }
+ else if (strlen(loadName) != MD5_LENGTH * 2 + 2)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Bad file name provided for cache to restore.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Bad file name provided for cache to restore.\n";
+
+ EnableSignals();
+
+ return NULL;
+ }
+
+ istream *cachefs = NULL;
+ char md5String[(MD5_LENGTH * 2) + 2];
+ md5_byte_t md5FromFile[MD5_LENGTH];
+
+ char *cacheName = new char[strlen(loadPath) + strlen(loadName) + 3];
+
+ strcpy(cacheName, loadPath);
+ strcat(cacheName, "/");
+ strcat(cacheName, loadName);
+
+ #ifdef TEST
+ *logofs << "Proxy: Name of cache file is '"
+ << cacheName << "'.\n" << logofs_flush;
+ #endif
+
+ cachefs = new ifstream(cacheName, ios::in | ios::binary);
+
+ unsigned char version[4];
+
+ if (cachefs == NULL || GetData(cachefs, version, 4) < 0)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Can't read cache file '"
+ << cacheName << "'.\n" << logofs_flush;;
+ #endif
+
+ handleFailOnLoad(cacheName, "A");
+
+ delete cachefs;
+
+ delete [] cacheName;
+
+ EnableSignals();
+
+ return NULL;
+ }
+
+ int major;
+ int minor;
+ int patch;
+
+ if (handleLoadVersion(version, major, minor, patch) < 0)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: WARNING! Incompatible version '"
+ << major << "." << minor << "." << patch
+ << "' in cache file '" << cacheName
+ << "'.\n" << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Incompatible version '"
+ << major << "." << minor << "." << patch
+ << "' in cache file '" << cacheName
+ << "'.\n" << logofs_flush;
+
+ if (control -> ProxyMode == proxy_server)
+ {
+ handleFailOnLoad(cacheName, "B");
+ }
+ else
+ {
+ //
+ // Simply remove the cache file.
+ //
+
+ unlink(cacheName);
+ }
+
+ delete cachefs;
+
+ delete [] cacheName;
+
+ EnableSignals();
+
+ return NULL;
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Reading from cache file version '"
+ << major << "." << minor << "." << patch
+ << "'.\n" << logofs_flush;
+ #endif
+
+ if (GetData(cachefs, md5FromFile, MD5_LENGTH) < 0)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! No checksum in cache file '"
+ << loadName << "'.\n" << logofs_flush;
+ #endif
+
+ handleFailOnLoad(cacheName, "C");
+
+ delete cachefs;
+
+ delete [] cacheName;
+
+ EnableSignals();
+
+ return NULL;
+ }
+
+ md5_state_t *md5StateStream = NULL;
+ md5_byte_t *md5DigestStream = NULL;
+
+ md5StateStream = new md5_state_t();
+ md5DigestStream = new md5_byte_t[MD5_LENGTH];
+
+ md5_init(md5StateStream);
+
+ //
+ // Use the virtual method of the proxy class.
+ //
+
+ if (handleLoadAllStores(cachefs, md5StateStream) < 0)
+ {
+ handleFailOnLoad(cacheName, "D");
+
+ delete cachefs;
+
+ delete md5StateStream;
+ delete [] md5DigestStream;
+
+ delete [] cacheName;
+
+ EnableSignals();
+
+ return NULL;
+ }
+
+ md5_append(md5StateStream, (const md5_byte_t *) loadName, strlen(loadName));
+ md5_finish(md5StateStream, md5DigestStream);
+
+ for (int i = 0; i < MD5_LENGTH; i++)
+ {
+ if (md5DigestStream[i] != md5FromFile[i])
+ {
+ #ifdef PANIC
+
+ *logofs << "Proxy: PANIC! Bad checksum for cache file '"
+ << cacheName << "'.\n" << logofs_flush;
+
+ for (unsigned int i = 0; i < MD5_LENGTH; i++)
+ {
+ sprintf(md5String + (i * 2), "%02X", md5FromFile[i]);
+ }
+
+ *logofs << "Proxy: PANIC! Saved checksum is '"
+ << md5String << "'.\n" << logofs_flush;
+
+ for (unsigned int i = 0; i < MD5_LENGTH; i++)
+ {
+ sprintf(md5String + (i * 2),"%02X", md5DigestStream[i]);
+ }
+
+ *logofs << "Proxy: PANIC! Calculated checksum is '"
+ << md5String << "'.\n" << logofs_flush;
+
+ #endif
+
+ handleFailOnLoad(cacheName, "E");
+
+ delete cachefs;
+
+ delete md5StateStream;
+ delete [] md5DigestStream;
+
+ delete [] cacheName;
+
+ EnableSignals();
+
+ return NULL;
+ }
+ }
+
+ delete cachefs;
+
+ delete md5StateStream;
+ delete [] md5DigestStream;
+
+ delete [] cacheName;
+
+ //
+ // Restore the original handlers.
+ //
+
+ EnableSignals();
+
+ #ifdef TEST
+ *logofs << "Proxy: Successfully loaded cache file '"
+ << loadName << "'.\n" << logofs_flush;
+ #endif
+
+ //
+ // Return the string provided by caller.
+ //
+
+ return loadName;
+}
+
+int Proxy::allocateChannelMap(int fd)
+{
+ //
+ // We assume that the fd is lower than
+ // the maximum allowed number. This is
+ // checked at the time the connection
+ // is accepted.
+ //
+
+ if (fd < 0 || fd >= CONNECTIONS_LIMIT)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Internal error allocating "
+ << "new channel with FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Internal error allocating "
+ << "new channel with FD#" << fd_ << ".\n";
+
+ HandleCleanup();
+ }
+
+ for (int channelId = 0;
+ channelId < CONNECTIONS_LIMIT;
+ channelId++)
+ {
+ if (checkLocalChannelMap(channelId) == 1 &&
+ fdMap_[channelId] == -1)
+ {
+ fdMap_[channelId] = fd;
+ channelMap_[fd] = channelId;
+
+ #ifdef TEST
+ *logofs << "Proxy: Allocated new channel ID#"
+ << channelId << " with FD#" << fd << ".\n"
+ << logofs_flush;
+ #endif
+
+ return channelId;
+ }
+ }
+
+ //
+ // No available channel is remaining.
+ //
+
+ #ifdef TEST
+ *logofs << "Proxy: WARNING! Can't allocate a new channel "
+ << "for FD#" << fd_ << ".\n" << logofs_flush;
+ #endif
+
+ return -1;
+}
+
+int Proxy::checkChannelMap(int channelId)
+{
+ //
+ // To be acceptable, the channel id must
+ // be an id that is not possible to use
+ // to allocate channels at this side.
+ //
+
+ if (checkLocalChannelMap(channelId) == 1)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Can't open a new channel "
+ << "with invalid ID#" << channelId << ".\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Can't open a new channel "
+ << "with invalid ID#" << channelId << ".\n";
+
+ return -1;
+ }
+ else if (channels_[channelId] != NULL)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Can't open a new channel "
+ << "over an existing ID#" << channelId
+ << " with FD#" << getFd(channelId)
+ << ".\n" << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Can't open a new channel "
+ << "over an existing ID#" << channelId
+ << " with FD#" << getFd(channelId)
+ << ".\n";
+
+ return -1;
+ }
+
+ return 1;
+}
+
+int Proxy::assignChannelMap(int channelId, int fd)
+{
+ //
+ // We assume that the fd is lower than
+ // the maximum allowed number. This is
+ // checked at the time the connection
+ // is accepted.
+ //
+
+ if (channelId < 0 || channelId >= CONNECTIONS_LIMIT ||
+ fd < 0 || fd >= CONNECTIONS_LIMIT)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Internal error assigning "
+ << "new channel with FD#" << fd_ << ".\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Internal error assigning "
+ << "new channel with FD#" << fd_ << ".\n";
+
+ HandleCleanup();
+ }
+
+ fdMap_[channelId] = fd;
+ channelMap_[fd] = channelId;
+
+ return 1;
+}
+
+void Proxy::cleanupChannelMap(int channelId)
+{
+ int fd = fdMap_[channelId];
+
+ if (fd != -1)
+ {
+ fdMap_[channelId] = -1;
+ channelMap_[fd] = -1;
+ }
+}
+
+int Proxy::addControlCodes(T_proxy_code code, int data)
+{
+ //
+ // Flush the encode buffer plus all the outstanding
+ // control codes if there is not enough space for
+ // the new control message. We need to ensure that
+ // there are further bytes available, in the case
+ // we will need to add more token control messages.
+ //
+
+ if (controlLength_ + 3 > CONTROL_CODES_THRESHOLD)
+ {
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Flushing control messages "
+ << "while sending code '" << DumpControl(code)
+ << "'.\n" << logofs_flush;
+ #endif
+
+ if (handleFlush() < 0)
+ {
+ return -1;
+ }
+ }
+
+ controlCodes_[controlLength_++] = 0;
+ controlCodes_[controlLength_++] = (unsigned char) code;
+ controlCodes_[controlLength_++] = (unsigned char) (data == -1 ? 0 : data);
+
+ //
+ // Account for the control frame.
+ //
+
+ statistics -> addFrameOut();
+
+ return 1;
+}
+
+void Proxy::setSplitTimeout(int channelId)
+{
+ int needed = channels_[channelId] -> needSplit();
+
+ if (needed != isTimestamp(timeouts_.splitTs))
+ {
+ if (needed == 1)
+ {
+ #if defined(TEST) || defined(INFO) || defined(SPLIT)
+ *logofs << "Proxy: SPLIT! Allocating split timestamp at "
+ << strMsTimestamp() << ".\n" << logofs_flush;
+ #endif
+
+ timeouts_.splitTs = getTimestamp();
+ }
+ else
+ {
+ T_list &channelList = activeChannels_.getList();
+
+ for (T_list::iterator j = channelList.begin();
+ j != channelList.end(); j++)
+ {
+ int channelId = *j;
+
+ if (channels_[channelId] != NULL &&
+ channels_[channelId] -> needSplit() == 1)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: SPLIT! Channel for FD#"
+ << getFd(channelId) << " still needs splits.\n"
+ << logofs_flush;
+ #endif
+
+ return;
+ }
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(SPLIT)
+ *logofs << "Proxy: SPLIT! Resetting split timestamp at "
+ << strMsTimestamp() << ".\n" << logofs_flush;
+ #endif
+
+ timeouts_.splitTs = nullTimestamp();
+ }
+ }
+}
+
+void Proxy::setMotionTimeout(int channelId)
+{
+ int needed = channels_[channelId] -> needMotion();
+
+ if (needed != isTimestamp(timeouts_.motionTs))
+ {
+ if (channels_[channelId] -> needMotion() == 1)
+ {
+ #if defined(TEST) || defined(INFO) || defined(SPLIT)
+ *logofs << "Proxy: Allocating motion timestamp at "
+ << strMsTimestamp() << ".\n" << logofs_flush;
+ #endif
+
+ timeouts_.motionTs = getTimestamp();
+ }
+ else
+ {
+ T_list &channelList = activeChannels_.getList();
+
+ for (T_list::iterator j = channelList.begin();
+ j != channelList.end(); j++)
+ {
+ int channelId = *j;
+
+ if (channels_[channelId] != NULL &&
+ channels_[channelId] -> needMotion() == 1)
+ {
+ #ifdef TEST
+ *logofs << "Proxy: SPLIT! Channel for FD#"
+ << getFd(channelId) << " still needs motions.\n"
+ << logofs_flush;
+ #endif
+
+ return;
+ }
+ }
+
+ #if defined(TEST) || defined(INFO) || defined(SPLIT)
+ *logofs << "Proxy: Resetting motion timestamp at "
+ << strMsTimestamp() << ".\n" << logofs_flush;
+ #endif
+
+ timeouts_.motionTs = nullTimestamp();
+ }
+ }
+}
+
+void Proxy::increaseChannels(int channelId)
+{
+ #ifdef TEST
+ *logofs << "Proxy: Adding channel " << channelId
+ << " to the list of active channels.\n"
+ << logofs_flush;
+ #endif
+
+ activeChannels_.add(channelId);
+
+ #ifdef TEST
+ *logofs << "Proxy: There are " << activeChannels_.getSize()
+ << " allocated channels for proxy FD#" << fd_
+ << ".\n" << logofs_flush;
+ #endif
+}
+
+void Proxy::decreaseChannels(int channelId)
+{
+ #ifdef TEST
+ *logofs << "Proxy: Removing channel " << channelId
+ << " from the list of active channels.\n"
+ << logofs_flush;
+ #endif
+
+ activeChannels_.remove(channelId);
+
+ #ifdef TEST
+ *logofs << "Proxy: There are " << activeChannels_.getSize()
+ << " allocated channels for proxy FD#" << fd_
+ << ".\n" << logofs_flush;
+ #endif
+}
+
+int Proxy::allocateTransport(int channelFd, int channelId)
+{
+ if (transports_[channelId] == NULL)
+ {
+ transports_[channelId] = new Transport(channelFd);
+
+ if (transports_[channelId] == NULL)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Can't allocate transport for "
+ << "channel id " << channelId << ".\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Can't allocate transport for "
+ << "channel id " << channelId << ".\n";
+
+ return -1;
+ }
+ }
+ else if (transports_[channelId] ->
+ getType() != transport_agent)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Transport for channel id "
+ << channelId << " should be null.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Transport for channel id "
+ << channelId << " should be null.\n";
+
+ return -1;
+ }
+
+ return 1;
+}
+
+int Proxy::deallocateTransport(int channelId)
+{
+ //
+ // Transport for the agent connection
+ // is passed from the outside when
+ // creating the channel.
+ //
+
+ if (transports_[channelId] ->
+ getType() != transport_agent)
+ {
+ delete transports_[channelId];
+ }
+
+ transports_[channelId] = NULL;
+
+ return 1;
+}
+
+int Proxy::handleNewGenericConnection(int clientFd, T_channel_type type, const char *label)
+{
+ int channelId = allocateChannelMap(clientFd);
+
+ if (channelId == -1)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Maximum mumber of available "
+ << "channels exceeded.\n" << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Maximum mumber of available "
+ << "channels exceeded.\n";
+
+ return -1;
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Channel for " << label << " descriptor "
+ << "FD#" << clientFd << " mapped to ID#"
+ << channelId << ".\n"
+ << logofs_flush;
+ #endif
+
+ //
+ // Turn queuing off for path server-to-proxy.
+ //
+
+ SetNoDelay(clientFd, 1);
+
+ if (allocateTransport(clientFd, channelId) < 0)
+ {
+ return -1;
+ }
+
+ switch (type)
+ {
+ case channel_cups:
+ {
+ channels_[channelId] = new CupsChannel(transports_[channelId], compressor_);
+
+ break;
+ }
+ case channel_smb:
+ {
+ channels_[channelId] = new SmbChannel(transports_[channelId], compressor_);
+
+ break;
+ }
+ case channel_media:
+ {
+ channels_[channelId] = new MediaChannel(transports_[channelId], compressor_);
+
+ break;
+ }
+ case channel_http:
+ {
+ channels_[channelId] = new HttpChannel(transports_[channelId], compressor_);
+
+ break;
+ }
+ case channel_font:
+ {
+ channels_[channelId] = new FontChannel(transports_[channelId], compressor_);
+
+ break;
+ }
+ default:
+ {
+ channels_[channelId] = new SlaveChannel(transports_[channelId], compressor_);
+
+ break;
+ }
+ }
+
+ if (channels_[channelId] == NULL)
+ {
+ deallocateTransport(channelId);
+
+ return -1;
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Accepted new connection to "
+ << label << " server.\n" << logofs_flush;
+ #endif
+
+ cerr << "Info" << ": Accepted new connection to "
+ << label << " server.\n";
+
+ increaseChannels(channelId);
+
+ switch (type)
+ {
+ case channel_cups:
+ {
+ if (handleControl(code_new_cups_connection, channelId) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case channel_smb:
+ {
+ if (handleControl(code_new_smb_connection, channelId) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case channel_media:
+ {
+ if (handleControl(code_new_media_connection, channelId) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case channel_http:
+ {
+ if (handleControl(code_new_http_connection, channelId) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ case channel_font:
+ {
+ if (handleControl(code_new_font_connection, channelId) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ default:
+ {
+ if (handleControl(code_new_slave_connection, channelId) < 0)
+ {
+ return -1;
+ }
+
+ break;
+ }
+ }
+
+ channels_[channelId] -> handleConfiguration();
+
+ return 1;
+}
+
+int Proxy::handleNewSlaveConnection(int clientFd)
+{
+ if (control -> isProtoStep7() == 1)
+ {
+ return handleNewGenericConnection(clientFd, channel_slave, "slave");
+ }
+ else
+ {
+ #ifdef TEST
+ *logofs << "Proxy: WARNING! Not sending unsupported "
+ << "'code_new_slave_connection' message.\n"
+ << logofs_flush;
+ #endif
+
+ return -1;
+ }
+}
+
+int Proxy::handleNewGenericConnectionFromProxy(int channelId, T_channel_type type,
+ const char *hostname, int port, const char *label)
+{
+ if (port <= 0)
+ {
+ //
+ // This happens when user has disabled
+ // forwarding of the specific service.
+ //
+
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Refusing attempted connection "
+ << "to " << label << " server.\n" << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Refusing attempted connection "
+ << "to " << label << " server.\n";
+
+ return -1;
+ }
+
+ const char *serverHost = hostname;
+ int serverAddrFamily = AF_INET;
+ sockaddr *serverAddr = NULL;
+ unsigned int serverAddrLength = 0;
+
+ #ifdef TEST
+ *logofs << "Proxy: Connecting to " << label
+ << " server '" << serverHost << "' on TCP port '"
+ << port << "'.\n" << logofs_flush;
+ #endif
+
+ int serverIPAddr = GetHostAddress(serverHost);
+
+ if (serverIPAddr == 0)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Unknown " << label
+ << " server host '" << serverHost << "'.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Unknown " << label
+ << " server host '" << serverHost
+ << "'.\n";
+
+ return -1;
+ }
+
+ sockaddr_in *serverAddrTCP = new sockaddr_in;
+
+ serverAddrTCP -> sin_family = AF_INET;
+ serverAddrTCP -> sin_port = htons(port);
+ serverAddrTCP -> sin_addr.s_addr = serverIPAddr;
+
+ serverAddr = (sockaddr *) serverAddrTCP;
+ serverAddrLength = sizeof(sockaddr_in);
+
+ //
+ // Connect to the requested server.
+ //
+
+ int serverFd = socket(serverAddrFamily, SOCK_STREAM, PF_UNSPEC);
+
+ if (serverFd < 0)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Call to socket failed. "
+ << "Error is " << EGET() << " '" << ESTR()
+ << "'.\n" << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Call to socket failed. "
+ << "Error is " << EGET() << " '" << ESTR()
+ << "'.\n";
+
+ delete serverAddrTCP;
+
+ return -1;
+ }
+ else if (connect(serverFd, serverAddr, serverAddrLength) < 0)
+ {
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Connection to " << label
+ << " server '" << serverHost << ":" << port
+ << "' failed with error '" << ESTR() << "'.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Connection to " << label
+ << " server '" << serverHost << ":" << port
+ << "' failed with error '" << ESTR() << "'.\n";
+
+ close(serverFd);
+
+ delete serverAddrTCP;
+
+ return -1;
+ }
+
+ delete serverAddrTCP;
+
+ if (handlePostConnectionFromProxy(channelId, serverFd, type, label) < 0)
+ {
+ return -1;
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Forwarded new connection to "
+ << label << " server on port '" << port
+ << "'.\n" << logofs_flush;
+ #endif
+
+ cerr << "Info" << ": Forwarded new connection to "
+ << label << " server on port '" << port
+ << "'.\n";
+
+ return 1;
+}
+
+int Proxy::handleNewGenericConnectionFromProxy(int channelId, T_channel_type type,
+ const char *hostname, const char *path, const char *label)
+{
+ if (path == NULL || *path == '\0' )
+ {
+ //
+ // This happens when user has disabled
+ // forwarding of the specific service.
+ //
+
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Refusing attempted connection "
+ << "to " << label << " server.\n" << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Refusing attempted connection "
+ << "to " << label << " server.\n";
+
+ return -1;
+ }
+
+ sockaddr_un serverAddrUnix;
+
+ unsigned int serverAddrLength = sizeof(sockaddr_un);
+
+ int serverAddrFamily = AF_UNIX;
+
+ serverAddrUnix.sun_family = AF_UNIX;
+
+ const int serverAddrNameLength = 108;
+
+ strncpy(serverAddrUnix.sun_path, path, serverAddrNameLength);
+
+ *(serverAddrUnix.sun_path + serverAddrNameLength - 1) = '\0';
+
+ #ifdef TEST
+ *logofs << "Proxy: Connecting to " << label << " server "
+ << "on Unix port '" << path << "'.\n" << logofs_flush;
+ #endif
+
+ //
+ // Connect to the requested server.
+ //
+
+ int serverFd = socket(serverAddrFamily, SOCK_STREAM, PF_UNSPEC);
+
+ if (serverFd < 0)
+ {
+ #ifdef PANIC
+ *logofs << "Proxy: PANIC! Call to socket failed. "
+ << "Error is " << EGET() << " '" << ESTR()
+ << "'.\n" << logofs_flush;
+ #endif
+
+ cerr << "Error" << ": Call to socket failed. "
+ << "Error is " << EGET() << " '" << ESTR()
+ << "'.\n";
+
+ return -1;
+ }
+ else if (connect(serverFd, (sockaddr *) &serverAddrUnix, serverAddrLength) < 0)
+ {
+ #ifdef WARNING
+ *logofs << "Proxy: WARNING! Connection to " << label
+ << " server on Unix port '" << path << "' failed "
+ << "with error " << EGET() << ", '" << ESTR() << "'.\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Connection to " << label
+ << " server on Unix port '" << path << "' failed "
+ << "with error " << EGET() << ", '" << ESTR() << "'.\n";
+
+ close(serverFd);
+
+ return -1;
+ }
+
+ if (handlePostConnectionFromProxy(channelId, serverFd, type, label) < 0)
+ {
+ return -1;
+ }
+
+ #ifdef TEST
+ *logofs << "Proxy: Forwarded new connection to "
+ << label << " server on Unix port '" << path
+ << "'.\n" << logofs_flush;
+ #endif
+
+ cerr << "Info" << ": Forwarded new connection to "
+ << label << " server on Unix port '" << path
+ << "'.\n";
+
+ return 1;
+}
+
+int Proxy::handleNewSlaveConnectionFromProxy(int channelId)
+{
+ //
+ // Implementation is incomplete. Opening a
+ // slave channel should let the proxy fork
+ // a new client and pass to it the channel
+ // descriptors. For now we make the channel
+ // fail immediately.
+ //
+ // #include <fcntl.h>
+ // #include <sys/types.h>
+ // #include <sys/stat.h>
+ //
+ // char *slaveServer = "/dev/null";
+ //
+ // #ifdef TEST
+ // *logofs << "Proxy: Opening file '" << slaveServer
+ // << "'.\n" << logofs_flush;
+ // #endif
+ //
+ // int serverFd = open(slaveServer, O_RDWR);
+ //
+ // if (handlePostConnectionFromProxy(channelId, serverFd, channel_slave, "slave") < 0)
+ // {
+ // return -1;
+ // }
+ //
+
+ #ifdef WARNING
+ *logofs << "Proxy: Refusing new slave connection for "
+ << "channel ID#" << channelId << "\n"
+ << logofs_flush;
+ #endif
+
+ cerr << "Warning" << ": Refusing new slave connection for "
+ << "channel ID#" << channelId << "\n";
+
+ return -1;
+}
+
+int Proxy::handlePostConnectionFromProxy(int channelId, int serverFd,
+ T_channel_type type, const char *label)
+{
+ //
+ // Turn queuing off for path proxy-to-server.
+ //
+
+ SetNoDelay(serverFd, 1);
+
+ assignChannelMap(channelId, serverFd);
+
+ #ifdef TEST
+ *logofs << "Proxy: Descriptor FD#" << serverFd
+ << " mapped to channel ID#" << channelId << ".\n"
+ << logofs_flush;
+ #endif
+
+ if (allocateTransport(serverFd, channelId) < 0)
+ {
+ return -1;
+ }
+
+ switch (type)
+ {
+ case channel_cups:
+ {
+ channels_[channelId] = new CupsChannel(transports_[channelId], compressor_);
+ break;
+ }
+ case channel_smb:
+ {
+ channels_[channelId] = new SmbChannel(transports_[channelId], compressor_);
+
+ break;
+ }
+ case channel_media:
+ {
+ channels_[channelId] = new MediaChannel(transports_[channelId], compressor_);
+
+ break;
+ }
+ case channel_http:
+ {
+ channels_[channelId] = new HttpChannel(transports_[channelId], compressor_);
+
+ break;
+ }
+ case channel_font:
+ {
+ channels_[channelId] = new FontChannel(transports_[channelId], compressor_);
+
+ break;
+ }
+ default:
+ {
+ channels_[channelId] = new SlaveChannel(transports_[channelId], compressor_);
+
+ break;
+ }
+ }
+
+ if (channels_[channelId] == NULL)
+ {
+ deallocateTransport(channelId);
+
+ return -1;
+ }
+
+ increaseChannels(channelId);
+
+ channels_[channelId] -> handleConfiguration();
+
+ return 1;
+}