X-Git-Url: https://git.netwichtig.de/gitweb/?a=blobdiff_plain;f=src%2Fmodules%2Fextra%2Fm_ziplink.cpp;h=976a27b5c7293bec55c4a336b428aaec1f3d989d;hb=6d03943426dcce76ba66567a9b18425a5ebb4c0c;hp=faa5ecdbd9fcb3ee69963908ed87f45ea5bb5473;hpb=66b53f9ef83b7099a3acc0dc071db44201188b37;p=user%2Fhenk%2Fcode%2Finspircd.git diff --git a/src/modules/extra/m_ziplink.cpp b/src/modules/extra/m_ziplink.cpp index faa5ecdbd..976a27b5c 100644 --- a/src/modules/extra/m_ziplink.cpp +++ b/src/modules/extra/m_ziplink.cpp @@ -1,259 +1,149 @@ -#include -#include - -#include "zlib.h" - -#include "inspircd_config.h" -#include "configreader.h" -#include "users.h" -#include "channels.h" -#include "modules.h" +/* +------------------------------------+ + * | Inspire Internet Relay Chat Daemon | + * +------------------------------------+ + * + * InspIRCd: (C) 2002-2009 InspIRCd Development Team + * See: http://wiki.inspircd.org/Credits + * + * This program is free but copyrighted software; see + * the file COPYING for details. + * + * --------------------------------------------------- + */ -#include "socket.h" -#include "hashcomp.h" #include "inspircd.h" - +#include #include "transport.h" +#include /* $ModDesc: Provides zlib link support for servers */ /* $LinkerFlags: -lz */ /* $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 + * a flush after each chunk. A frame may contain multiple lines * and should be treated as raw binary data. - * */ -static InspIRCd* SI; - -enum izip_status { IZIP_OPEN, IZIP_CLOSED }; - -const unsigned int CHUNK = 16384; - -class CountedBuffer : public classbase -{ - int bufptr; /* Current tail location */ - unsigned char* buffer; /* Current buffer contents */ - int bufsz; /* Current buffer size */ - int amount_expected; /* Amount of data expected */ - int amount_read; /* Amount of data read so far */ - public: - CountedBuffer() - { - bufsz = 1024; - buffer = new unsigned char[bufsz + 1]; - bufptr = 0; - amount_read = 0; - amount_expected = 0; - } +/* Status of a connection */ +enum izip_status { IZIP_CLOSED = 0, IZIP_OPEN }; - ~CountedBuffer() - { - delete[] buffer; - } - - void AddData(unsigned char* data, int data_length) - { - SI->Log(DEBUG,"AddData, %d bytes to add", data_length); - if ((data_length + bufptr) > bufsz) - { - SI->Log(DEBUG,"Need to extend buffer to %d, is now %d", data_length + bufptr, bufsz); - /* Buffer is too small, enlarge it and copy contents */ - int old_bufsz = bufsz; - unsigned char* temp = buffer; - - bufsz += data_length; - buffer = new unsigned char[bufsz + 1]; - - memcpy(buffer, temp, old_bufsz); - - delete[] temp; - } - - SI->Log(DEBUG,"Copy data in at pos %d", bufptr); - - memcpy(buffer + bufptr, data, data_length); - bufptr += data_length; - amount_read += data_length; - - SI->Log(DEBUG,"Amount read is now %d, bufptr is now %d", amount_read, bufptr); - - if ((!amount_expected) && (amount_read >= 4)) - { - SI->Log(DEBUG,"We dont yet have an expected amount"); - /* We have enough to read an int */ - int* size = (int*)buffer; - amount_expected = ntohl(*size); - SI->Log(DEBUG,"Expected amount is %d", amount_expected); - } - } - - 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 ((amount_read - 4) >= amount_expected) - { - SI->Log(DEBUG,"We have enough for the frame (have %d)", (amount_read - 4)); - int amt_ex = amount_expected; - /* Yes, we have enough now */ - memcpy(frame, buffer + 4, amount_expected > maxsize ? maxsize : amount_expected); - RemoveFirstFrame(); - return (amt_ex > maxsize) ? maxsize : amt_ex; - } - } - /* Not enough for a frame yet, COME AGAIN! */ - return 0; - } - - void RemoveFirstFrame() - { - SI->Log(DEBUG,"Removing first frame from buffer sized %d", amount_expected); - unsigned char* temp = buffer; - - bufsz -= (amount_expected + 4); - buffer = new unsigned char[bufsz + 1]; - - SI->Log(DEBUG,"Shrunk buffer to %d", bufsz); - - memcpy(buffer, temp + amount_expected + 4, bufsz); - - amount_read -= (amount_expected + 4); - SI->Log(DEBUG,"Amount read now %d", amount_read); - - if (amount_read >= 4) - { - /* We have enough to read an int */ - int* size = (int*)buffer; - amount_expected = ntohl(*size); - } - else - amount_expected = 0; - - SI->Log(DEBUG,"Amount expected now %d", amount_expected); - - bufptr = 0; - - delete[] temp; - } -}; - -/** 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; /* uncompress stream */ + izip_status status; /* Connection status */ + std::string outbuf; /* Holds output buffer (compressed) */ + std::string inbuf; /* Holds input buffer (compressed) */ }; 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; - + + /* Used for reading data from the wire and compressing data to. */ + char *net_buffer; + unsigned int net_buffer_size; public: - - ModuleZLib(InspIRCd* Me) - : Module::Module(Me) - { - ServerInstance->PublishInterface("InspSocketHook", this); + + ModuleZLib() + { + ServerInstance->Modules->PublishInterface("BufferedSocketHook", this); + + sessions = new izip_session[ServerInstance->SE->GetMaxFds()]; + for (int i = 0; i < ServerInstance->SE->GetMaxFds(); i++) + sessions[i].status = IZIP_CLOSED; total_out_compressed = total_in_compressed = 0; - total_out_uncompressed = total_out_uncompressed = 0; + total_out_uncompressed = total_in_uncompressed = 0; + Implementation eventlist[] = { I_OnStats, I_OnRequest }; + ServerInstance->Modules->Attach(eventlist, this, 2); - SI = ServerInstance; + // Allocate a buffer which is used for reading and writing data + net_buffer_size = ServerInstance->Config->NetBufferSize; + net_buffer = new char[net_buffer_size]; } - virtual ~ModuleZLib() + ~ModuleZLib() { + ServerInstance->Modules->UnpublishInterface("BufferedSocketHook", this); + delete[] sessions; + delete[] net_buffer; } - virtual Version GetVersion() + Version GetVersion() { - return Version(1, 1, 0, 0, VF_VENDOR, API_VERSION); + return Version("Provides zlib link support for servers", 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 */ + 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"; - try - { - ret = ServerInstance->Config->AddIOHook((Module*)this, (InspSocket*)ISR->Sock) ? (char*)"OK" : NULL; - } - catch (ModuleException& e) - { - return NULL; - } - return ret; + ISR->Sock->AddIOHook(this); + return "OK"; } else if (strcmp("IS_UNHOOK", request->GetId()) == 0) { - return ServerInstance->Config->DelIOHook((InspSocket*)ISR->Sock) ? (char*)"OK" : NULL; + ISR->Sock->DelIOHook(); + return "OK"; } 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) */ + ModResult OnStats(char symbol, User* user, string_list &results) { if (symbol == 'z') { std::string sn = ServerInstance->Config->ServerName; - 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); + /* 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 = (total_out_compressed / (total_out_uncompressed + 0.001)) * 100; + float inbound_r = (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); + float total_r = (total_compressed / (total_uncompressed + 0.001)) * 100; char outbound_ratio[MAXBUF], inbound_ratio[MAXBUF], combined_ratio[MAXBUF]; @@ -265,25 +155,28 @@ class ModuleZLib : public Module 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; + results.push_back(sn+" 304 "+user->nick+" :ZIPSTATS percentage_of_original_outbound_traffic = "+outbound_ratio); + results.push_back(sn+" 304 "+user->nick+" :ZIPSTATS percentage_of_orignal_inbound_traffic = "+inbound_ratio); + results.push_back(sn+" 304 "+user->nick+" :ZIPSTATS total_size_of_original_traffic = "+combined_ratio); + return MOD_RES_PASSTHRU; } - return 0; + return MOD_RES_PASSTHRU; } - virtual void OnRawSocketAccept(int fd, const std::string &ip, int localport) + void OnStreamSocketConnect(StreamSocket* user) { + OnStreamSocketAccept(user, 0, 0); + } + + void OnRawSocketAccept(StreamSocket* user, irc::sockets::sockaddrs*, irc::sockets::sockaddrs*) + { + int fd = user->GetFd(); + izip_session* session = &sessions[fd]; - - /* allocate deflate state */ - session->fd = fd; - session->status = IZIP_OPEN; - session->inbuf = new CountedBuffer(); - ServerInstance->Log(DEBUG,"session->inbuf ALLOC = %d, %08x", fd, session->inbuf); + /* Just in case... */ + session->outbuf.clear(); session->c_stream.zalloc = (alloc_func)0; session->c_stream.zfree = (free_func)0; @@ -292,164 +185,228 @@ class ModuleZLib : public Module session->d_stream.zalloc = (alloc_func)0; session->d_stream.zfree = (free_func)0; session->d_stream.opaque = (voidpf)0; - } - virtual void OnRawSocketConnect(int fd) - { - OnRawSocketAccept(fd, "", 0); + /* If we cant call this, well, we're boned. */ + if (inflateInit(&session->d_stream) != Z_OK) + { + session->status = IZIP_CLOSED; + return; + } + + /* Same here */ + if (deflateInit(&session->c_stream, Z_BEST_COMPRESSION) != Z_OK) + { + inflateEnd(&session->d_stream); + session->status = IZIP_CLOSED; + return; + } + + /* Just in case, do this last */ + session->status = IZIP_OPEN; } - virtual void OnRawSocketClose(int fd) + void OnStreamSocketClose(StreamSocket* user) { + int fd = user->GetFd(); CloseSession(&sessions[fd]); } - - virtual int OnRawSocketRead(int fd, char* buffer, unsigned int count, int &readresult) + + int OnStreamSocketRead(StreamSocket* user, std::string& recvq) { + int fd = user->GetFd(); + /* Find the sockets session */ izip_session* session = &sessions[fd]; if (session->status == IZIP_CLOSED) - return 1; - - unsigned char compr[CHUNK + 1]; - unsigned int total_decomp = 0; + return -1; - readresult = read(fd, compr, CHUNK); - - if (readresult > 0) + if (session->inbuf.empty()) { - session->inbuf->AddData(compr, readresult); - - int size = session->inbuf->GetFrame(compr, CHUNK); - while ((size) && (total_decomp < count)) + /* Read read_buffer_size bytes at a time to the buffer (usually 2.5k) */ + int readresult = read(fd, net_buffer, net_buffer_size); + + if (readresult < 0) { - - session->d_stream.next_in = (Bytef*)compr; - session->d_stream.avail_in = 0; - session->d_stream.next_out = (Bytef*)(buffer + total_decomp); - if (inflateInit(&session->d_stream) != Z_OK) - return -EBADF; - - 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; - } - - 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); - - size = session->inbuf->GetFrame(compr, CHUNK); + if (errno == EINTR || errno == EAGAIN) + return 0; } + if (readresult <= 0) + return -1; - buffer[total_decomp] = 0; + total_in_compressed += readresult; - ServerInstance->Log(DEBUG,"Complete buffer: '%s' size=%d", buffer, total_decomp); + /* Copy the compressed data into our input buffer */ + session->inbuf.append(net_buffer, readresult); } - return (readresult > 0); - } - virtual int OnRawSocketWrite(int fd, const char* buffer, int count) - { - ServerInstance->Log(DEBUG,"Compressing %d bytes", count); + size_t in_len = session->inbuf.length(); + char* buffer = ServerInstance->GetReadBuffer(); + int count = ServerInstance->Config->NetBufferSize; - izip_session* session = &sessions[fd]; - int ocount = count; + /* Prepare decompression */ + session->d_stream.next_in = (Bytef *)session->inbuf.c_str(); + session->d_stream.avail_in = in_len; - if (!count) + session->d_stream.next_out = (Bytef*)buffer; + /* Last byte is reserved for NULL terminating that beast */ + session->d_stream.avail_out = count - 1; + + /* Z_SYNC_FLUSH: Do as much as possible */ + int ret = inflate(&session->d_stream, Z_SYNC_FLUSH); + /* TODO CloseStream() in here at random places */ + switch (ret) { - ServerInstance->Log(DEBUG,"Nothing to do!"); - return 1; + case Z_NEED_DICT: + case Z_STREAM_ERROR: + /* This is one of the 'not supposed to happen' things. + * Memory corruption, anyone? + */ + Error(session, "General Error. This is not supposed to happen :/"); + break; + case Z_DATA_ERROR: + Error(session, "Decompression failed, malformed data"); + break; + case Z_MEM_ERROR: + Error(session, "Out of memory"); + break; + case Z_BUF_ERROR: + /* This one is non-fatal, buffer is just full + * (can't happen here). + */ + Error(session, "Internal error. This is not supposed to happen."); + break; + case Z_STREAM_END: + /* This module *never* generates these :/ */ + Error(session, "End-of-stream marker received"); + break; + case Z_OK: + break; + default: + /* NO WAI! This can't happen. All errors are handled above. */ + Error(session, "Unknown error"); + break; } - - if(session->status != IZIP_OPEN) + if (ret != Z_OK) { - CloseSession(session); - return 0; + return -1; } - unsigned char compr[count*2+4]; + /* Update the inbut buffer */ + unsigned int input_compressed = in_len - session->d_stream.avail_in; + session->inbuf = session->inbuf.substr(input_compressed); - if (deflateInit(&session->c_stream, Z_BEST_COMPRESSION) != Z_OK) - { - ServerInstance->Log(DEBUG,"Deflate init failed"); - } + /* Update counters (Old size - new size) */ + unsigned int uncompressed_length = (count - 1) - session->d_stream.avail_out; + total_in_uncompressed += uncompressed_length; + + /* Null-terminate the buffer -- this doesnt harm binary data */ + recvq.append(buffer, uncompressed_length); + return 1; + } + + int OnStreamSocketWrite(StreamSocket* user, std::string& sendq) + { + int fd = user->GetFd(); + izip_session* session = &sessions[fd]; - session->c_stream.next_in = (Bytef*)buffer; - session->c_stream.next_out = compr+4; + if(session->status != IZIP_OPEN) + /* Seriously, wtf? */ + return -1; - while ((session->c_stream.total_in < (unsigned int)count) && (session->c_stream.total_out < (unsigned int)count*2)) + int ret; + + /* This loop is really only supposed to run once, but in case 'compr' + * is filled up somehow we are prepared to handle this situation. + */ + unsigned int offset = 0; + do { - session->c_stream.avail_in = session->c_stream.avail_out = 1; /* force small buffers */ - if (deflate(&session->c_stream, Z_NO_FLUSH) != Z_OK) + /* Prepare compression */ + session->c_stream.next_in = (Bytef*)sendq.data() + offset; + session->c_stream.avail_in = sendq.length() - offset; + + session->c_stream.next_out = (Bytef*)net_buffer; + session->c_stream.avail_out = net_buffer_size; + + /* Compress the text */ + ret = deflate(&session->c_stream, Z_SYNC_FLUSH); + /* TODO CloseStream() in here at random places */ + switch (ret) { - ServerInstance->Log(DEBUG,"Couldnt deflate!"); - CloseSession(session); - return 0; + case Z_OK: + break; + case Z_BUF_ERROR: + /* This one is non-fatal, buffer is just full + * (can't happen here). + */ + Error(session, "Internal error. This is not supposed to happen."); + break; + case Z_STREAM_ERROR: + /* This is one of the 'not supposed to happen' things. + * Memory corruption, anyone? + */ + Error(session, "General Error. This is also not supposed to happen."); + break; + default: + Error(session, "Unknown error"); + break; } - } - /* 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; - } - deflateEnd(&session->c_stream); + if (ret != Z_OK) + return 0; - total_out_uncompressed += ocount; - total_out_compressed += session->c_stream.total_out; + /* Space before - space after stuff was added to this */ + unsigned int compressed = net_buffer_size - session->c_stream.avail_out; + unsigned int uncompressed = sendq.length() - session->c_stream.avail_in; - 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. - */ - memcpy(compr, &x, sizeof(x)); - write(fd, compr, session->c_stream.total_out+4); + /* Make it skip the data which was compressed already */ + offset += uncompressed; - return ocount; - } - - void CloseSession(izip_session* session) - { - if (session->status = IZIP_OPEN) + /* Update stats */ + total_out_uncompressed += uncompressed; + total_out_compressed += compressed; + + /* Add compressed to the output buffer */ + session->outbuf.append((const char*)net_buffer, compressed); + } while (session->c_stream.avail_in != 0); + + /* Lets see how much we can send out */ + 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) { - session->status = IZIP_CLOSED; - delete session->inbuf; + if (errno == EAGAIN) + return 0; + else + { + session->outbuf.clear(); + return -1; + } } - } - -}; -class ModuleZLibFactory : public ModuleFactory -{ - public: - ModuleZLibFactory() - { + return 1; } - - ~ModuleZLibFactory() + + void Error(izip_session* session, const std::string &text) { + ServerInstance->SNO->WriteToSnoMask('l', "ziplink error: " + text); } - - virtual Module * CreateModule(InspIRCd* Me) + + void CloseSession(izip_session* session) { - return new ModuleZLib(Me); + if (session->status == IZIP_OPEN) + { + session->status = IZIP_CLOSED; + session->outbuf.clear(); + inflateEnd(&session->d_stream); + deflateEnd(&session->c_stream); + } } + }; +MODULE_INIT(ModuleZLib) -extern "C" void * init_module( void ) -{ - return new ModuleZLibFactory; -}