src/modules/extra/m_ziplink.cpp

   1 #include <string>
   2 #include <vector>
   3
   4 #include "zlib.h"
   5
   6 #include "inspircd_config.h"
   7 #include "configreader.h"
   8 #include "users.h"
   9 #include "channels.h"
  10 #include "modules.h"
  11
  12 #include "socket.h"
  13 #include "hashcomp.h"
  14 #include "inspircd.h"
  15
  16 #include "transport.h"
  17
  18 /* $ModDesc: Provides zlib link support for servers */
  19 /* $LinkerFlags: -lz */
  20 /* $ModDep: transport.h */
  21
  22 /*
  23  * Compressed data is transmitted across the link in the following format:
  24  *
  25  *   0   1   2   3   4 ... n
  26  * +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
  27  * |       n       |              Z0 -> Zn                         |
  28  * +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
  29  *
  30  * Where: n is the size of a frame, in network byte order, 4 bytes.
  31  * Z0 through Zn are Zlib compressed data, n bytes in length.
  32  *
  33  * If the module fails to read the entire frame, then it will buffer
  34  * the portion of the last frame it received, then attempt to read
  35  * the next part of the frame next time a write notification arrives.
  36  *
  37  * ZLIB_BEST_COMPRESSION (9) is used for all sending of data with
  38  * a flush after each frame. A frame may contain multiple lines
  39  * and should be treated as raw binary data.
  40  *
  41  */
  42
  43
  44 enum izip_status { IZIP_WAITFIRST, IZIP_OPEN, IZIP_CLOSED };
  45
  46 const unsigned int CHUNK = 16384;
  47
  48 /** Represents an ZIP user's extra data
  49  */
  50 class izip_session : public classbase
  51 {
  52  public:
  53         z_stream c_stream; /* compression stream */
  54         z_stream d_stream; /* decompress stream */
  55         izip_status status;
  56         int need_bytes;
  57         int fd;
  58         std::string inbuf;
  59 };
  60
  61 class ModuleZLib : public Module
  62 {
  63         izip_session sessions[MAX_DESCRIPTORS];
  64         float total_out_compressed;
  65         float total_in_compressed;
  66         float total_out_uncompressed;
  67         float total_in_uncompressed;
  68
  69  public:
  70
  71         ModuleZLib(InspIRCd* Me)
  72                 : Module::Module(Me)
  73         {
  74                 ServerInstance->PublishInterface("InspSocketHook", this);
  75
  76                 total_out_compressed = total_in_compressed = 0;
  77                 total_out_uncompressed = total_out_uncompressed = 0;
  78         }
  79
  80         virtual ~ModuleZLib()
  81         {
  82         }
  83
  84         virtual Version GetVersion()
  85         {
  86                 return Version(1, 1, 0, 0, VF_VENDOR, API_VERSION);
  87         }
  88
  89         void Implements(char* List)
  90         {
  91                 List[I_OnRawSocketConnect] = List[I_OnRawSocketAccept] = List[I_OnRawSocketClose] = List[I_OnRawSocketRead] = List[I_OnRawSocketWrite] = 1;
  92                 List[I_OnStats] = List[I_OnRequest] = 1;
  93         }
  94
  95         virtual char* OnRequest(Request* request)
  96         {
  97                 ISHRequest* ISR = (ISHRequest*)request;
  98                 if (strcmp("IS_NAME", request->GetId()) == 0)
  99                 {
 100                         return "zip";
 101                 }
 102                 else if (strcmp("IS_HOOK", request->GetId()) == 0)
 103                 {
 104                         return ServerInstance->Config->AddIOHook((Module*)this, (InspSocket*)ISR->Sock) ? (char*)"OK" : NULL;
 105                 }
 106                 else if (strcmp("IS_UNHOOK", request->GetId()) == 0)
 107                 {
 108                         return ServerInstance->Config->DelIOHook((InspSocket*)ISR->Sock) ? (char*)"OK" : NULL;
 109                 }
 110                 else if (strcmp("IS_HSDONE", request->GetId()) == 0)
 111                 {
 112                         return "OK";
 113                 }
 114                 else if (strcmp("IS_ATTACH", request->GetId()) == 0)
 115                 {
 116                         return NULL;
 117                 }
 118                 return NULL;
 119         }
 120
 121         virtual int OnStats(char symbol, userrec* user, string_list &results)
 122         {
 123                 if (symbol == 'C')
 124                 {
 125                         std::string sn = ServerInstance->Config->ServerName;
 126
 127                         float outbound_r = 100 - ((total_out_compressed / (total_out_uncompressed + 0.001)) * 100);
 128                         float inbound_r = 100 - ((total_in_compressed / (total_in_uncompressed + 0.001)) * 100);
 129
 130                         float total_compressed = total_in_compressed + total_out_compressed;
 131                         float total_uncompressed = total_in_uncompressed + total_out_uncompressed;
 132
 133                         float total_r = 100 - ((total_compressed / (total_uncompressed + 0.001)) * 100);
 134
 135                         char outbound_ratio[MAXBUF], inbound_ratio[MAXBUF], combined_ratio[MAXBUF];
 136
 137                         sprintf(outbound_ratio, "%3.2f%%", outbound_r);
 138                         sprintf(inbound_ratio, "%3.2f%%", inbound_r);
 139                         sprintf(combined_ratio, "%3.2f%%", total_r);
 140
 141                         results.push_back(sn+" 304 "+user->nick+" : ZIPSTATS outbound_compressed   = "+ConvToStr(total_out_compressed));
 142                         results.push_back(sn+" 304 "+user->nick+" : ZIPSTATS inbound_compressed    = "+ConvToStr(total_in_compressed));
 143                         results.push_back(sn+" 304 "+user->nick+" : ZIPSTATS outbound_uncompressed = "+ConvToStr(total_out_uncompressed));
 144                         results.push_back(sn+" 304 "+user->nick+" : ZIPSTATS inbound_uncompressed  = "+ConvToStr(total_in_uncompressed));
 145                         results.push_back(sn+" 304 "+user->nick+" : ZIPSTATS ----------------------------");
 146                         results.push_back(sn+" 304 "+user->nick+" : ZIPSTATS OUTBOUND RATIO        = "+outbound_ratio);
 147                         results.push_back(sn+" 304 "+user->nick+" : ZIPSTATS INBOUND RATIO         = "+inbound_ratio);
 148                         results.push_back(sn+" 304 "+user->nick+" : ZIPSTATS COMBINED RATIO        = "+combined_ratio);
 149                         return 0;
 150                 }
 151
 152                 return 0;
 153         }
 154
 155         virtual void OnRawSocketAccept(int fd, const std::string &ip, int localport)
 156         {
 157                 izip_session* session = &sessions[fd];
 158
 159                 /* allocate deflate state */
 160                 session->fd = fd;
 161                 session->status = IZIP_WAITFIRST;
 162
 163                 session->need_bytes = 0;
 164
 165                 session->c_stream.zalloc = (alloc_func)0;
 166                 session->c_stream.zfree = (free_func)0;
 167                 session->c_stream.opaque = (voidpf)0;
 168
 169                 session->d_stream.zalloc = (alloc_func)0;
 170                 session->d_stream.zfree = (free_func)0;
 171                 session->d_stream.opaque = (voidpf)0;
 172
 173         }
 174
 175         virtual void OnRawSocketConnect(int fd)
 176         {
 177                 OnRawSocketAccept(fd, "", 0);
 178         }
 179
 180         virtual void OnRawSocketClose(int fd)
 181         {
 182                 CloseSession(&sessions[fd]);
 183         }
 184
 185         virtual int OnRawSocketRead(int fd, char* buffer, unsigned int count, int &readresult)
 186         {
 187                 izip_session* session = &sessions[fd];
 188
 189                 if (session->status == IZIP_CLOSED)
 190                         return 1;
 191
 192                 int size = 0;
 193
 194                 if (session->need_bytes)
 195                 {
 196                         size = session->need_bytes;
 197                 }
 198                 else
 199                 {
 200                         if (read(fd, &size, sizeof(size)) != sizeof(size))
 201                                 return 0;
 202                         size = ntohl(size);
 203                 }
 204
 205                 ServerInstance->Log(DEBUG,"Size of frame to read: %d%s", size, session->need_bytes ? " (remainder of last frame)" : "");
 206
 207                 unsigned char compr[size+1+session->need_bytes];
 208
 209                 readresult = read(fd, compr + session->need_bytes, size);
 210
 211                 if (readresult == size)
 212                 {
 213                         if(session->status == IZIP_WAITFIRST)
 214                         {
 215                                 session->status = IZIP_OPEN;
 216                         }
 217
 218                         /* Reassemble first part of last frame */
 219                         if (session->need_bytes)
 220                         {
 221                                 for (size_t i = 0; i < session->inbuf.length(); i++)
 222                                         compr[i] = session->inbuf[i];
 223                         }
 224
 225                         session->d_stream.next_in  = (Bytef*)compr;
 226                         session->d_stream.avail_in = 0;
 227                         session->d_stream.next_out = (Bytef*)buffer;
 228                         if (inflateInit(&session->d_stream) != Z_OK)
 229                                 return -EBADF;
 230                         session->status = IZIP_OPEN;
 231
 232                         while ((session->d_stream.total_out < count) && (session->d_stream.total_in < (unsigned int)readresult))
 233                         {
 234                                 session->d_stream.avail_in = session->d_stream.avail_out = 1; /* force small buffers */
 235                                 if (inflate(&session->d_stream, Z_NO_FLUSH) == Z_STREAM_END)
 236                                         break;
 237                         }
 238
 239                         inflateEnd(&session->d_stream);
 240
 241                         total_in_compressed += readresult;
 242                         readresult = session->d_stream.total_out;
 243                         total_in_uncompressed += session->d_stream.total_out;
 244
 245                         buffer[readresult] = 0;
 246                         session->need_bytes = 0;
 247                 }
 248                 else
 249                 {
 250                         /* We need to buffer here */
 251                         ServerInstance->Log(DEBUG,"Didnt read whole frame, got %d bytes of %d!", readresult, size);
 252                         session->need_bytes = ((readresult > -1) ? (size - readresult) : (size));
 253                         if (readresult > 0)
 254                         {
 255                                 /* Do it this way because it needs to be binary safe */
 256                                 for (int i = 0; i < readresult; i++)
 257                                         session->inbuf += compr[i];
 258                         }
 259                 }
 260
 261                 return (readresult > 0);
 262         }
 263
 264         virtual int OnRawSocketWrite(int fd, const char* buffer, int count)
 265         {
 266                 int ocount = count;
 267                 if (!count)
 268                 {
 269                         ServerInstance->Log(DEBUG,"Nothing to do!");
 270                         return 1;
 271                 }
 272
 273                 unsigned char compr[count*2+4];
 274
 275                 izip_session* session = &sessions[fd];
 276
 277                 if(session->status == IZIP_WAITFIRST)
 278                 {
 279                         session->status = IZIP_OPEN;
 280                 }
 281
 282                 if (deflateInit(&session->c_stream, Z_BEST_COMPRESSION) != Z_OK)
 283                 {
 284                         ServerInstance->Log(DEBUG,"Deflate init failed");
 285                 }
 286
 287                 if(session->status != IZIP_OPEN)
 288                 {
 289                         ServerInstance->Log(DEBUG,"State not open!");
 290                         CloseSession(session);
 291                         return 0;
 292                 }
 293
 294                 session->c_stream.next_in  = (Bytef*)buffer;
 295                 session->c_stream.next_out = compr+4;
 296
 297                 while ((session->c_stream.total_in < (unsigned int)count) && (session->c_stream.total_out < (unsigned int)count*2))
 298                 {
 299                         session->c_stream.avail_in = session->c_stream.avail_out = 1; /* force small buffers */
 300                         if (deflate(&session->c_stream, Z_NO_FLUSH) != Z_OK)
 301                         {
 302                                 ServerInstance->Log(DEBUG,"Couldnt deflate!");
 303                                 CloseSession(session);
 304                                 return 0;
 305                         }
 306                 }
 307                 /* Finish the stream, still forcing small buffers: */
 308                 for (;;)
 309                 {
 310                         session->c_stream.avail_out = 1;
 311                         if (deflate(&session->c_stream, Z_FINISH) == Z_STREAM_END)
 312                                 break;
 313                 }
 314
 315                 deflateEnd(&session->c_stream);
 316
 317                 total_out_uncompressed += ocount;
 318                 total_out_compressed += session->c_stream.total_out;
 319
 320                 int x = htonl(session->c_stream.total_out);
 321                 /** XXX: We memcpy it onto the start of the buffer like this to save ourselves a write().
 322                  * A memcpy of 4 or so bytes is less expensive and gives the tcp stack more chance of
 323                  * assembling the frame size into the same packet as the compressed frame.
 324                  */
 325                 memcpy(compr, &x, sizeof(x));
 326                 write(fd, compr, session->c_stream.total_out+4);
 327
 328                 return ocount;
 329         }
 330
 331         void CloseSession(izip_session* session)
 332         {
 333                 session->status = IZIP_CLOSED;
 334         }
 335
 336 };
 337
 338 class ModuleZLibFactory : public ModuleFactory
 339 {
 340  public:
 341         ModuleZLibFactory()
 342         {
 343         }
 344
 345         ~ModuleZLibFactory()
 346         {
 347         }
 348
 349         virtual Module * CreateModule(InspIRCd* Me)
 350         {
 351                 return new ModuleZLib(Me);
 352         }
 353 };
 354
 355
 356 extern "C" void * init_module( void )
 357 {
 358         return new ModuleZLibFactory;
 359 }