X-Git-Url: https://git.netwichtig.de/gitweb/?a=blobdiff_plain;f=src%2Fmodules%2Fextra%2Fm_ziplink.cpp;h=7b1cb281c2c53651f559b02bedf9f7fb982c3566;hb=813bc55a8496875cef52e6da42d608e4d2fa35da;hp=59ae89605fd2cf69959e8f61e448ed004c098ca1;hpb=8d3c6acc8cb23eb87d00395a0bf399470b190f08;p=user%2Fhenk%2Fcode%2Finspircd.git diff --git a/src/modules/extra/m_ziplink.cpp b/src/modules/extra/m_ziplink.cpp index 59ae89605..7b1cb281c 100644 --- a/src/modules/extra/m_ziplink.cpp +++ b/src/modules/extra/m_ziplink.cpp @@ -1,119 +1,284 @@ -#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 "ssl.h" +#include "transport.h" /* $ModDesc: Provides zlib link support for servers */ /* $LinkerFlags: -lz */ -/* $ModDep: ssl.h */ +/* $ModDep: transport.h */ + +/* + * Compressed data is transmitted across the link in the following format: + * + * 0 1 2 3 4 ... n + * +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+ + * | n | Z0 -> Zn | + * +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+ + * + * Where: n is the size of a frame, in network byte order, 4 bytes. + * Z0 through Zn are Zlib compressed data, n bytes in length. + * + * If the module fails to read the entire frame, then it will buffer + * the portion of the last frame it received, then attempt to read + * the next part of the frame next time a write notification arrives. + * + * ZLIB_BEST_COMPRESSION (9) is used for all sending of data with + * a flush after each frame. A frame may contain multiple lines + * and should be treated as raw binary data. + * + */ + +/* 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::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) + { + buffer.append((const char*)data, data_length); + this->NextFrameSize(); + } -enum izip_status { IZIP_WAITFIRST, IZIP_OPEN, IZIP_CLOSED }; + /** Works out the size of the next compressed frame + */ + void NextFrameSize() + { + if ((!amount_expected) && (buffer.length() >= 4)) + { + /* 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); + } + } -const unsigned int CHUNK = 16384; + /** 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) + { + /* We know how much we're expecting... + * Do we have enough yet? + */ + if (buffer.length() >= amount_expected) + { + int j = 0; + for (unsigned int i = 0; i < amount_expected; i++, j++) + frame[i] = buffer[i]; + + buffer = buffer.substr(j); + amount_expected = 0; + NextFrameSize(); + return j; + } + } + /* Not enough for a frame yet, COME AGAIN! */ + return 0; + } +}; -/** 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; + 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; + float total_in_uncompressed; public: 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; + 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_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) { - return ServerInstance->Config->AddIOHook((Module*)this, (InspSocket*)ISR->Sock) ? (char*)"OK" : NULL; + /* Attach to an inspsocket */ + const char* ret = "OK"; + try + { + ret = ServerInstance->Config->AddIOHook((Module*)this, (BufferedSocket*)ISR->Sock) ? "OK" : NULL; + } + catch (ModuleException& e) + { + return NULL; + } + return ret; } 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; } + /* 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); + + float total_compressed = total_in_compressed + total_out_compressed; + float total_uncompressed = total_in_uncompressed + total_out_uncompressed; + + float total_r = 100 - ((total_compressed / (total_uncompressed + 0.001)) * 100); + + char outbound_ratio[MAXBUF], inbound_ratio[MAXBUF], combined_ratio[MAXBUF]; + + sprintf(outbound_ratio, "%3.2f%%", outbound_r); + sprintf(inbound_ratio, "%3.2f%%", inbound_r); + sprintf(combined_ratio, "%3.2f%%", total_r); + + results.push_back(sn+" 304 "+user->nick+" :ZIPSTATS outbound_compressed = "+ConvToStr(total_out_compressed)); + results.push_back(sn+" 304 "+user->nick+" :ZIPSTATS inbound_compressed = "+ConvToStr(total_in_compressed)); + results.push_back(sn+" 304 "+user->nick+" :ZIPSTATS outbound_uncompressed = "+ConvToStr(total_out_uncompressed)); + results.push_back(sn+" 304 "+user->nick+" :ZIPSTATS inbound_uncompressed = "+ConvToStr(total_in_uncompressed)); + results.push_back(sn+" 304 "+user->nick+" :ZIPSTATS outbound_ratio = "+outbound_ratio); + results.push_back(sn+" 304 "+user->nick+" :ZIPSTATS inbound_ratio = "+inbound_ratio); + results.push_back(sn+" 304 "+user->nick+" :ZIPSTATS combined_ratio = "+combined_ratio); + return 0; + } + + return 0; + } virtual void OnRawSocketAccept(int fd, const std::string &ip, int localport) { izip_session* session = &sessions[fd]; - /* allocate deflate state */ + /* allocate state and buffers */ session->fd = fd; - session->status = IZIP_WAITFIRST; + session->status = IZIP_OPEN; + session->inbuf = new CountedBuffer(); session->c_stream.zalloc = (alloc_func)0; session->c_stream.zfree = (free_func)0; session->c_stream.opaque = (voidpf)0; - if (deflateInit(&session->c_stream, Z_DEFAULT_COMPRESSION) != Z_OK) - return; - session->d_stream.zalloc = (alloc_func)0; session->d_stream.zfree = (free_func)0; session->d_stream.opaque = (voidpf)0; - - if (deflateInit(&session->d_stream, Z_DEFAULT_COMPRESSION) != Z_OK) - return; } virtual void OnRawSocketConnect(int fd) { + /* Nothing special needs doing here compared to accept() */ OnRawSocketAccept(fd, "", 0); } @@ -121,160 +286,167 @@ 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; - - int size = 0; - if (read(fd, &size, sizeof(size)) != sizeof(size)) return 0; - size = ntohl(size); - - ServerInstance->Log(DEBUG,"Size of frame to read: %d", size); + unsigned char compr[CHUNK + 4]; + unsigned int offset = 0; + unsigned int total_size = 0; - unsigned char compr[size+1]; + /* Read CHUNK bytes at a time to the buffer (usually 128k) */ + readresult = read(fd, compr, CHUNK); - readresult = read(fd, compr, size); - - if (readresult == size) + /* Did we get anything? */ + if (readresult > 0) { - if(session->status == IZIP_WAITFIRST) - { - session->status = IZIP_OPEN; - } - - session->d_stream.next_in = (Bytef*)compr; - session->d_stream.avail_in = 0; - session->d_stream.next_out = (Bytef*)buffer; - if (inflateInit(&session->d_stream) != Z_OK) - return -EBADF; - session->status = IZIP_OPEN; - - while ((session->d_stream.total_out < count) && (session->d_stream.total_in < (unsigned int)readresult)) + /* Add it to the frame queue */ + session->inbuf->AddData(compr, readresult); + total_in_compressed += readresult; + + /* Parse all completed frames */ + int size = 0; + while ((size = session->inbuf->GetFrame(compr, CHUNK)) != 0) { - session->d_stream.avail_in = session->d_stream.avail_out = 1; /* force small buffers */ - if (inflate(&session->d_stream, Z_NO_FLUSH) == Z_STREAM_END) - break; - } + session->d_stream.next_in = (Bytef*)compr; + session->d_stream.avail_in = 0; + session->d_stream.next_out = (Bytef*)(buffer + offset); - inflateEnd(&session->d_stream); + /* If we cant call this, well, we're boned. */ + if (inflateInit(&session->d_stream) != Z_OK) + return 0; + + while ((session->d_stream.total_out < count) && (session->d_stream.total_in < (unsigned int)size)) + { + session->d_stream.avail_in = session->d_stream.avail_out = 1; + if (inflate(&session->d_stream, Z_NO_FLUSH) == Z_STREAM_END) + break; + } + + /* Stick a fork in me, i'm done */ + inflateEnd(&session->d_stream); - readresult = session->d_stream.total_out; + /* 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[readresult] = 0; + /* Null-terminate the buffer -- this doesnt harm binary data */ + buffer[total_size] = 0; - ServerInstance->Log(DEBUG,"DECOMPRESSED: '%s'", buffer); - } - else - { - /* XXX: We need to buffer here, really. */ - if (readresult == -1) - { - ServerInstance->Log(DEBUG,"Error: %s", strerror(errno)); - if (errno == EAGAIN) - ServerInstance->Log(DEBUG,"(EAGAIN)"); - } - ServerInstance->Log(DEBUG,"Didnt read whole frame, got %d bytes of %d!", readresult, size); - } + /* Set the read size to the correct total size */ + readresult = total_size; + } return (readresult > 0); } virtual int OnRawSocketWrite(int fd, const char* buffer, int count) { + izip_session* session = &sessions[fd]; int ocount = count; - ServerInstance->Log(DEBUG,"Write event of %d uncompressed bytes: '%s'", count, buffer); - - if (!count) - { - ServerInstance->Log(DEBUG,"Nothing to do!"); - return 1; - } - unsigned char compr[count*2+4]; - - izip_session* session = &sessions[fd]; + if (!count) /* Nothing to do! */ + return 0; - if(session->status == IZIP_WAITFIRST) + if(session->status != IZIP_OPEN) { - session->status = IZIP_OPEN; + /* Seriously, wtf? */ + CloseSession(session); + return 0; } - // Z_BEST_COMPRESSION - if (deflateInit(&session->c_stream, Z_BEST_COMPRESSION) != Z_OK) - { - ServerInstance->Log(DEBUG,"Deflate init failed"); - } + unsigned char compr[CHUNK + 4]; - if(session->status != IZIP_OPEN) + /* Gentlemen, start your engines! */ + if (deflateInit(&session->c_stream, Z_BEST_COMPRESSION) != Z_OK) { - ServerInstance->Log(DEBUG,"State not open!"); 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; - while ((session->c_stream.total_in < (unsigned int)count) && (session->c_stream.total_out < (unsigned int)count*2)) + /* 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); - ServerInstance->Log(DEBUG,"Write %d compressed bytes", session->c_stream.total_out); - int x = htonl(session->c_stream.total_out); - memcpy(compr, &x, sizeof(x)); - write(fd, compr, session->c_stream.total_out+4); + total_out_uncompressed += ocount; + total_out_compressed += session->c_stream.total_out; - deflateEnd(&session->c_stream); + /** 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. + */ + session->outbuf.append((const char*)compr, session->c_stream.total_out + 4); + + /* 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) { - session->status = IZIP_CLOSED; + 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; -}