X-Git-Url: https://git.netwichtig.de/gitweb/?a=blobdiff_plain;f=src%2Fmodules%2Fextra%2Fm_ziplink.cpp;h=7b1cb281c2c53651f559b02bedf9f7fb982c3566;hb=813bc55a8496875cef52e6da42d608e4d2fa35da;hp=c2f197042528d4e708bacb0186e91477c860ab07;hpb=f83c5f620382dd6d415ce613e3cec9f92f693f70;p=user%2Fhenk%2Fcode%2Finspircd.git diff --git a/src/modules/extra/m_ziplink.cpp b/src/modules/extra/m_ziplink.cpp index c2f197042..7b1cb281c 100644 --- a/src/modules/extra/m_ziplink.cpp +++ b/src/modules/extra/m_ziplink.cpp @@ -1,18 +1,23 @@ -#include -#include - -#include "zlib.h" +/* +------------------------------------+ + * | Inspire Internet Relay Chat Daemon | + * +------------------------------------+ + * + * InspIRCd: (C) 2002-2008 InspIRCd Development Team + * See: http://www.inspircd.org/wiki/index.php/Credits + * + * This program is free but copyrighted software; see + * the file COPYING for details. + * + * --------------------------------------------------- + */ -#include "inspircd_config.h" -#include "configreader.h" +#include "inspircd.h" +#include #include "users.h" #include "channels.h" #include "modules.h" - #include "socket.h" #include "hashcomp.h" -#include "inspircd.h" - #include "transport.h" /* $ModDesc: Provides zlib link support for servers */ @@ -40,68 +45,75 @@ * */ -static InspIRCd* SI; - +/* Status of a connection */ enum izip_status { IZIP_OPEN, IZIP_CLOSED }; +/* Maximum transfer size per read operation */ const unsigned int CHUNK = 128 * 1024; +/* This class manages a compressed chunk of data preceeded by + * a length count. + * + * It can handle having multiple chunks of data in the buffer + * at any time. + */ class CountedBuffer : public classbase { - std::deque buffer; /* Current buffer contents */ - unsigned int amount_expected; /* Amount of data expected */ + std::string buffer; /* Current buffer contents */ + unsigned int amount_expected; /* Amount of data expected */ public: CountedBuffer() { amount_expected = 0; } + /** Adds arbitrary compressed data to the buffer. + * - Binsry safe, of course. + */ void AddData(unsigned char* data, int data_length) { - SI->Log(DEBUG,"AddData, %d bytes to add", data_length); - for (int i = 0; i < data_length; i++) - buffer.push_back(data[i]); - + buffer.append((const char*)data, data_length); this->NextFrameSize(); } + /** Works out the size of the next compressed frame + */ void NextFrameSize() { - if ((!amount_expected) && (buffer.size() >= 4)) + if ((!amount_expected) && (buffer.length() >= 4)) { - SI->Log(DEBUG,"We dont yet have an expected amount"); - /* We have enough to read an int */ - char sz[4]; - for (int i = 0; i < 4; i++) - { - sz[i] = buffer.front(); - buffer.pop_front(); - } - int* size = (int*)sz; - amount_expected = ntohl(*size); - SI->Log(DEBUG,"Expected amount is %d", amount_expected); + /* We have enough to read an int - + * Yes, this is safe, but its ugly. Give me + * a nicer way to read 4 bytes from a binary + * stream, and push them into a 32 bit int, + * and i'll consider replacing this. + */ + amount_expected = ntohl((buffer[3] << 24) | (buffer[2] << 16) | (buffer[1] << 8) | buffer[0]); + buffer = buffer.substr(4); } } + /** Gets the next frame and returns its size, or returns + * zero if there isnt one available yet. + * A frame can contain multiple plaintext lines. + * - Binary safe. + */ int GetFrame(unsigned char* frame, int maxsize) { if (amount_expected) { - SI->Log(DEBUG,"Were expecting a frame of size %d", amount_expected); /* We know how much we're expecting... * Do we have enough yet? */ - if (buffer.size() >= amount_expected) + if (buffer.length() >= amount_expected) { int j = 0; for (unsigned int i = 0; i < amount_expected; i++, j++) - { - frame[i] = buffer.front(); - buffer.pop_front(); - } + frame[i] = buffer[i]; + buffer = buffer.substr(j); + amount_expected = 0; NextFrameSize(); - return j; } } @@ -110,21 +122,24 @@ class CountedBuffer : public classbase } }; -/** Represents an ZIP user's extra data +/** Represents an zipped connections extra data */ class izip_session : public classbase { public: - z_stream c_stream; /* compression stream */ - z_stream d_stream; /* decompress stream */ - izip_status status; - int fd; - CountedBuffer* inbuf; + z_stream c_stream; /* compression stream */ + z_stream d_stream; /* decompress stream */ + izip_status status; /* Connection status */ + int fd; /* File descriptor */ + CountedBuffer* inbuf; /* Holds input buffer */ + std::string outbuf; /* Holds output buffer */ }; class ModuleZLib : public Module { - izip_session sessions[MAX_DESCRIPTORS]; + izip_session* sessions; + + /* Used for stats z extensions */ float total_out_compressed; float total_in_compressed; float total_out_uncompressed; @@ -135,42 +150,44 @@ class ModuleZLib : public Module ModuleZLib(InspIRCd* Me) : Module::Module(Me) { - ServerInstance->PublishInterface("InspSocketHook", this); + ServerInstance->Modules->PublishInterface("BufferedSocketHook", this); + + sessions = new izip_session[ServerInstance->SE->GetMaxFds()]; total_out_compressed = total_in_compressed = 0; total_out_uncompressed = total_out_uncompressed = 0; - - SI = ServerInstance; + Implementation eventlist[] = { I_OnRawSocketConnect, I_OnRawSocketAccept, I_OnRawSocketClose, I_OnRawSocketRead, I_OnRawSocketWrite, I_OnStats, I_OnRequest }; + ServerInstance->Modules->Attach(eventlist, this, 7); } virtual ~ModuleZLib() { + ServerInstance->Modules->UnpublishInterface("BufferedSocketHook", this); + delete[] sessions; } virtual Version GetVersion() { - return Version(1, 1, 0, 0, VF_VENDOR, API_VERSION); + return Version(1, 2, 0, 0, VF_VENDOR, API_VERSION); } - void Implements(char* List) - { - List[I_OnRawSocketConnect] = List[I_OnRawSocketAccept] = List[I_OnRawSocketClose] = List[I_OnRawSocketRead] = List[I_OnRawSocketWrite] = 1; - List[I_OnStats] = List[I_OnRequest] = 1; - } - virtual char* OnRequest(Request* request) + /* Handle BufferedSocketHook API requests */ + virtual const char* OnRequest(Request* request) { ISHRequest* ISR = (ISHRequest*)request; if (strcmp("IS_NAME", request->GetId()) == 0) { + /* Return name */ return "zip"; } else if (strcmp("IS_HOOK", request->GetId()) == 0) { - char* ret = "OK"; + /* Attach to an inspsocket */ + const char* ret = "OK"; try { - ret = ServerInstance->Config->AddIOHook((Module*)this, (InspSocket*)ISR->Sock) ? (char*)"OK" : NULL; + ret = ServerInstance->Config->AddIOHook((Module*)this, (BufferedSocket*)ISR->Sock) ? "OK" : NULL; } catch (ModuleException& e) { @@ -180,25 +197,40 @@ class ModuleZLib : public Module } else if (strcmp("IS_UNHOOK", request->GetId()) == 0) { - return ServerInstance->Config->DelIOHook((InspSocket*)ISR->Sock) ? (char*)"OK" : NULL; + /* Detatch from an inspsocket */ + return ServerInstance->Config->DelIOHook((BufferedSocket*)ISR->Sock) ? "OK" : NULL; } else if (strcmp("IS_HSDONE", request->GetId()) == 0) { + /* Check for completion of handshake + * (actually, this module doesnt handshake) + */ return "OK"; } else if (strcmp("IS_ATTACH", request->GetId()) == 0) { + /* Attach certificate data to the inspsocket + * (this module doesnt do that, either) + */ return NULL; } return NULL; } - virtual int OnStats(char symbol, userrec* user, string_list &results) + /* Handle stats z (misc stats) */ + virtual int OnStats(char symbol, User* user, string_list &results) { if (symbol == 'z') { std::string sn = ServerInstance->Config->ServerName; + /* Yeah yeah, i know, floats are ew. + * We used them here because we'd be casting to float anyway to do this maths, + * and also only floating point numbers can deal with the pretty large numbers + * involved in the total throughput of a server over a large period of time. + * (we dont count 64 bit ints because not all systems have 64 bit ints, and floats + * can still hold more. + */ float outbound_r = 100 - ((total_out_compressed / (total_out_uncompressed + 0.001)) * 100); float inbound_r = 100 - ((total_in_compressed / (total_in_uncompressed + 0.001)) * 100); @@ -230,12 +262,10 @@ class ModuleZLib : public Module { izip_session* session = &sessions[fd]; - /* allocate deflate state */ + /* allocate state and buffers */ session->fd = fd; session->status = IZIP_OPEN; - session->inbuf = new CountedBuffer(); - ServerInstance->Log(DEBUG,"session->inbuf ALLOC = %d, %08x", fd, session->inbuf); session->c_stream.zalloc = (alloc_func)0; session->c_stream.zfree = (free_func)0; @@ -248,6 +278,7 @@ class ModuleZLib : public Module virtual void OnRawSocketConnect(int fd) { + /* Nothing special needs doing here compared to accept() */ OnRawSocketAccept(fd, "", 0); } @@ -255,32 +286,40 @@ class ModuleZLib : public Module { CloseSession(&sessions[fd]); } - + virtual int OnRawSocketRead(int fd, char* buffer, unsigned int count, int &readresult) { + /* Find the sockets session */ izip_session* session = &sessions[fd]; if (session->status == IZIP_CLOSED) - return 1; + return 0; - unsigned char compr[CHUNK + 1]; - unsigned int total_decomp = 0; + unsigned char compr[CHUNK + 4]; + unsigned int offset = 0; + unsigned int total_size = 0; + /* Read CHUNK bytes at a time to the buffer (usually 128k) */ readresult = read(fd, compr, CHUNK); + /* Did we get anything? */ if (readresult > 0) { + /* Add it to the frame queue */ session->inbuf->AddData(compr, readresult); + total_in_compressed += readresult; - int size = session->inbuf->GetFrame(compr, CHUNK); - while ((size) && (total_decomp < count)) + /* Parse all completed frames */ + int size = 0; + while ((size = session->inbuf->GetFrame(compr, CHUNK)) != 0) { - session->d_stream.next_in = (Bytef*)compr; session->d_stream.avail_in = 0; - session->d_stream.next_out = (Bytef*)(buffer + total_decomp); + session->d_stream.next_out = (Bytef*)(buffer + offset); + + /* If we cant call this, well, we're boned. */ if (inflateInit(&session->d_stream) != Z_OK) - return -EBADF; + return 0; while ((session->d_stream.total_out < count) && (session->d_stream.total_in < (unsigned int)size)) { @@ -289,122 +328,125 @@ class ModuleZLib : public Module break; } + /* Stick a fork in me, i'm done */ inflateEnd(&session->d_stream); - - total_in_compressed += readresult; - readresult = session->d_stream.total_out; - total_in_uncompressed += session->d_stream.total_out; - - total_decomp += session->d_stream.total_out; - - ServerInstance->Log(DEBUG,"Decompressed %d bytes, total_decomp=%d: '%s'", session->d_stream.total_out, total_decomp, buffer); - if (total_decomp < count) - size = session->inbuf->GetFrame(compr, CHUNK); + /* Update counters and offsets */ + total_size += session->d_stream.total_out; + total_in_uncompressed += session->d_stream.total_out; + offset += session->d_stream.total_out; } - buffer[total_decomp] = 0; + /* Null-terminate the buffer -- this doesnt harm binary data */ + buffer[total_size] = 0; + + /* Set the read size to the correct total size */ + readresult = total_size; - ServerInstance->Log(DEBUG,"Complete buffer: '%s' size=%d", buffer, total_decomp); } return (readresult > 0); } virtual int OnRawSocketWrite(int fd, const char* buffer, int count) { - ServerInstance->Log(DEBUG,"Compressing %d bytes", count); - izip_session* session = &sessions[fd]; int ocount = count; - if (!count) - { - ServerInstance->Log(DEBUG,"Nothing to do!"); - return 1; - } + if (!count) /* Nothing to do! */ + return 0; if(session->status != IZIP_OPEN) { + /* Seriously, wtf? */ CloseSession(session); return 0; } - unsigned char compr[CHUNK]; + unsigned char compr[CHUNK + 4]; + /* Gentlemen, start your engines! */ if (deflateInit(&session->c_stream, Z_BEST_COMPRESSION) != Z_OK) { - ServerInstance->Log(DEBUG,"Deflate init failed"); + CloseSession(session); + return 0; } + /* Set buffer sizes (we reserve 4 bytes at the start of the + * buffer for the length counters) + */ session->c_stream.next_in = (Bytef*)buffer; - session->c_stream.next_out = compr+4; + session->c_stream.next_out = compr + 4; + /* Compress the text */ while ((session->c_stream.total_in < (unsigned int)count) && (session->c_stream.total_out < CHUNK)) { - session->c_stream.avail_in = session->c_stream.avail_out = 1; /* force small buffers */ + session->c_stream.avail_in = session->c_stream.avail_out = 1; if (deflate(&session->c_stream, Z_NO_FLUSH) != Z_OK) { - ServerInstance->Log(DEBUG,"Couldnt deflate!"); CloseSession(session); return 0; } } - /* Finish the stream, still forcing small buffers: */ - for (;;) - { - session->c_stream.avail_out = 1; - if (deflate(&session->c_stream, Z_FINISH) == Z_STREAM_END) - break; - } - + /* Finish the stream */ + for (session->c_stream.avail_out = 1; deflate(&session->c_stream, Z_FINISH) != Z_STREAM_END; session->c_stream.avail_out = 1); deflateEnd(&session->c_stream); total_out_uncompressed += ocount; total_out_compressed += session->c_stream.total_out; - int x = htonl(session->c_stream.total_out); - /** XXX: We memcpy it onto the start of the buffer like this to save ourselves a write(). - * A memcpy of 4 or so bytes is less expensive and gives the tcp stack more chance of - * assembling the frame size into the same packet as the compressed frame. + /** Assemble the frame length onto the frame, in network byte order */ + compr[0] = (session->c_stream.total_out >> 24); + compr[1] = (session->c_stream.total_out >> 16); + compr[2] = (session->c_stream.total_out >> 8); + compr[3] = (session->c_stream.total_out & 0xFF); + + /* Add compressed data plus leading length to the output buffer - + * Note, we may have incomplete half-sent frames in here. */ - memcpy(compr, &x, sizeof(x)); - write(fd, compr, session->c_stream.total_out+4); + session->outbuf.append((const char*)compr, session->c_stream.total_out + 4); - ServerInstance->Log(DEBUG,"Sending frame of size %d", x); + /* Lets see how much we can send out */ + int ret = write(fd, session->outbuf.data(), session->outbuf.length()); + /* Check for errors, and advance the buffer if any was sent */ + if (ret > 0) + session->outbuf = session->outbuf.substr(ret); + else if (ret < 1) + { + if (ret == -1) + { + if (errno == EAGAIN) + return 0; + else + { + session->outbuf.clear(); + return 0; + } + } + else + { + session->outbuf.clear(); + return 0; + } + } + + /* ALL LIES the lot of it, we havent really written + * this amount, but the layer above doesnt need to know. + */ return ocount; } void CloseSession(izip_session* session) { - if (session->status = IZIP_OPEN) + if (session->status == IZIP_OPEN) { session->status = IZIP_CLOSED; + session->outbuf.clear(); delete session->inbuf; } } }; -class ModuleZLibFactory : public ModuleFactory -{ - public: - ModuleZLibFactory() - { - } - - ~ModuleZLibFactory() - { - } - - virtual Module * CreateModule(InspIRCd* Me) - { - return new ModuleZLib(Me); - } -}; - +MODULE_INIT(ModuleZLib) -extern "C" void * init_module( void ) -{ - return new ModuleZLibFactory; -}