libs/neuronet/master.c

   1 /* master.c
   2  * Module : Master
   3  */
   4
   5 /*-------------------- Extern Headers Including --------------------*/
   6 #ifndef WIN32
   7
   8 #else /* WIN32 */
   9
  10 #include <windows.h> /* WSAStartup WSACleanup */
  11 #define MSG_DONTWAIT 0
  12
  13 #endif /* WIN32 */
  14
  15 #include <string.h> /* memset */
  16 #include <errno.h> /* errno */
  17
  18 #include <neuro/NEURO.h>
  19
  20 /*-------------------- Local Headers Including ---------------------*/
  21 #include <global.h>
  22 #include "common.h"
  23
  24 #include <neuro/nnet/epoll.h>
  25 #include "slave.h"
  26 #include "client.h"
  27 #include "server.h"
  28 #include "status.h"
  29 #include "util.h"
  30 #include "lbuf.h"
  31
  32 /*-------------------- Main Module Header --------------------------*/
  33 #include "master.h"
  34
  35 /*--------------------      Other       ----------------------------*/
  36
  37 NEURO_MODULE_CHANNEL("netmaster");
  38
  39 /*-------------------- Global Variables ----------------------------*/
  40
  41 /*-------------------- Static Variables ----------------------------*/
  42
  43 /*-------------------- Static Prototypes ---------------------------*/
  44
  45
  46
  47 /*-------------------- Static Functions ----------------------------*/
  48
  49 static void
  50 clean_statuses_elem(void *src)
  51 {
  52         Status *sta;
  53
  54         sta = src;
  55
  56         if (!sta)
  57                 return;
  58
  59         Status_Clear(sta);
  60 }
  61
  62 static void
  63 poll_disconnect_clients(Master *msr)
  64 {
  65         Slave **tmp = NULL;
  66
  67         if (Neuro_LBufIsEmpty(msr->disco_clients))
  68                 return;
  69
  70         Neuro_ResetLBuf(msr->disco_clients);
  71
  72         while ((tmp = Neuro_GiveNextLBuf(msr->disco_clients)))
  73         {
  74                 TRACE(Neuro_s("Disconnecting client %x from the buffer %x count %d", *tmp, tmp, Neuro_GiveLBufCount(msr->disco_clients)));
  75
  76                 if (msr->type == TYPE_SERVER)
  77                         Server_DisconnectClient(*tmp);
  78                 else
  79                         Client_Disconnect(*tmp);
  80
  81                 Neuro_SCleanLBuf(msr->disco_clients, tmp);
  82         }
  83 }
  84
  85 /* add a client to disconnect */
  86 static void
  87 add_client_disconnect(Slave *slv)
  88 {
  89         Slave **tmp = NULL;
  90
  91         if (!Neuro_LBufIsEmpty(slv->master->disco_clients))
  92         {
  93                 TRACE("Checking the disconnection buffer for duplicate entries");
  94
  95                 Neuro_ResetLBuf(slv->master->disco_clients);
  96
  97                 while ((tmp = Neuro_GiveNextLBuf(slv->master->disco_clients)))
  98                 {
  99                         TRACE(Neuro_s("is %x the same as %x -- %s", *tmp, slv, *tmp == slv ? "yes" : "no"));
 100
 101                         if (*tmp == slv)
 102                         {
 103                                 ERROR(Neuro_s("Slave %x already in the disconnection buffer!", slv));
 104
 105                                 return;
 106                         }
 107                 }
 108         }
 109
 110         Neuro_AllocLBuf(slv->master->disco_clients, sizeof(Slave**));
 111
 112         tmp = Neuro_GiveCurLBuf(slv->master->disco_clients);
 113
 114         *tmp = slv;
 115
 116         TRACE(Neuro_s("Adding client %x for disconnection buffer %x -- buffer total %d", slv, tmp, Neuro_GiveLBufCount(slv->master->disco_clients)));
 117
 118         /* so we no longer get any events for this client */
 119         Master_RmUfds(slv->master, slv);
 120 }
 121
 122 /*-------------------- Global Functions ----------------------------*/
 123
 124 void
 125 Master_AddUfds(Master *msr, Slave *slv)
 126 {
 127         EPOLL_EVENT event;
 128
 129         memset(&event, 0, sizeof(EPOLL_EVENT));
 130
 131         event.events = EPOLLIN | EPOLLPRI | EPOLLERR;
 132         event.data.ptr = slv;
 133
 134         Epoll_Ctl(msr->ep, EPOLL_CTL_ADD, slv->socket, &event);
 135 }
 136
 137 void
 138 Master_RmUfds(Master *msr, Slave *slv)
 139 {
 140         EPOLL_EVENT event;
 141
 142         memset(&event, 0, sizeof(EPOLL_EVENT));
 143
 144         Epoll_Ctl(msr->ep, EPOLL_CTL_DEL, slv->socket, &event);
 145 }
 146
 147 void
 148 Master_EditEvent(Event *event, Slave *slave, int sigmask)
 149 {
 150         event->slave = slave;
 151         event->sigmask = sigmask;
 152 }
 153
 154 void
 155 Master_PushEvent(Master *msr, Slave *slave, int sigmask)
 156 {
 157         EBUF *ce;
 158         Event *tmp;
 159
 160         if (!msr)
 161                 return;
 162
 163         ce = msr->cevents;
 164
 165         Neuro_AllocEBuf(ce, sizeof(Event*), sizeof(Event));
 166
 167         tmp = Neuro_GiveCurEBuf(ce);
 168
 169         tmp->sigmask = sigmask;
 170         tmp->slave = slave;
 171 }
 172
 173 int
 174 Master_PollEvent(Master *msr)
 175 {
 176         int i = 0;
 177         EPOLL_EVENT *events;
 178
 179         events = Epoll_Wait(msr->ep, 0, &i);
 180
 181         if (i == 0)
 182                 return 0;
 183
 184         if (i == -1)
 185         {
 186                 ERROR(Neuro_s("Error raised by Epoll_Wait : returned the value %d", errno));
 187
 188                 return 1;
 189         }
 190
 191         while (i-- > 0)
 192         {
 193                 if ((events[i].events & EPOLLIN) == EPOLLIN || (events[i].events & EPOLLPRI) == EPOLLPRI)
 194                 {
 195                         Slave *slave = events[i].data.ptr;
 196                         /* NEURO_TRACE("Input Event Catched", NULL); */
 197
 198                         if (slave->type == TYPE_SERVER)
 199                         {
 200                                 Server_Poll(slave);
 201
 202                                 return 0;
 203                         }
 204                         else
 205                         {
 206                                 int _err = 0;
 207
 208                                 if (msr->protocolType == SOCK_DGRAM && slave->type == TYPE_SERVER)
 209                                         return 0;
 210
 211                                 _err = Client_PollRead(slave);
 212
 213                                 switch (_err)
 214                                 {
 215                                         case 0:
 216                                         {
 217                                                 Client_PopData(slave);
 218                                         }
 219                                         break;
 220
 221                                         case 1:
 222                                         {
 223                                                 WARN("TODO : Disconnect client -- Client_PollRead requested it");
 224                                                 if (msr->type == TYPE_SERVER)
 225                                                         Status_AddPriority(msr, State_ClientDisconnect, NULL, 0, slave);
 226                                                 else /* TYPE_CLIENT */
 227                                                         Status_AddPriority(msr, State_Disconnect, NULL, 0, slave);
 228                                         }
 229                                         break;
 230
 231                                         case 2:
 232                                         {
 233                                                 return 1;
 234                                         }
 235                                         break;
 236                                 }
 237                         }
 238                 }
 239                 else if ((events[i].events & EPOLLOUT) == EPOLLOUT)
 240                 {
 241                         /* NEURO_TRACE("Output event catched", NULL); */
 242                 }
 243                 else if ((events[i].events & EPOLLERR) == EPOLLERR || (events[i].events & EPOLLHUP) == EPOLLHUP)
 244                 {
 245                         WARN("Error Event Catched, Disconnecting client");
 246
 247                         if (msr->type == TYPE_SERVER)
 248                                 Status_Add(msr, State_ClientDisconnect, NULL, 0, events[i].data.ptr);
 249                         else /* TYPE_CLIENT */
 250                                 Status_AddPriority(msr, State_Disconnect, NULL, 0, events[i].data.ptr);
 251                 }
 252         }
 253         return 0;
 254 }
 255
 256 int
 257 Master_SetSendPacketSize(Master *msr)
 258 {
 259         if (!msr)
 260                 return -1;
 261
 262         if (msr->inclpacket_size == 0)
 263         {
 264                 msr->inclpacket_size = 1;
 265
 266                 return 1;
 267         }
 268         else
 269                 msr->inclpacket_size = 0;
 270
 271         return 0;
 272 }
 273
 274 void
 275 Master_SetQuitFlag(Master *msr)
 276 {
 277         if (!msr)
 278                 return;
 279
 280         msr->type = 2;
 281 }
 282
 283 /*-------------------- Poll ----------------------------------------*/
 284
 285 Status *
 286 Master_Poll(Master *msr)
 287 {
 288         if (!msr)
 289                 return NULL;
 290
 291         if (msr->type == 2)
 292         {
 293                 TRACE("End user application flagged the Master as quit, exiting");
 294
 295                 Status_Set(msr->status, State_Disconnect, NULL, 0, NULL);
 296
 297                 return msr->status;
 298         }
 299
 300         if (!msr->slave)
 301         {
 302                 ERROR("Attempted to poll without a properly initialized slave");
 303                 return NULL;
 304         }
 305
 306         if (msr->type == TYPE_SERVER)
 307         {
 308                 if (!msr->slave->cType.server)
 309                 {
 310                         ERROR("Attempted to poll without a properly initialized server");
 311                         return NULL;
 312                 }
 313         }
 314         else if (msr->type == TYPE_CLIENT)
 315         {
 316                 if (!msr->slave->cType.client)
 317                 {
 318                         ERROR("Attempted to poll without a properly initialized client");
 319                         return NULL;
 320                 }
 321         }
 322         else
 323         {
 324                 ERROR("Invalid Master type, bailing out");
 325
 326                 return NULL;
 327         }
 328
 329         if (!Neuro_LBufIsEmpty(msr->statuses))
 330         {
 331                 /* delete the first entry of msr->statuses as it was used up */
 332         }
 333
 334
 335         /* disconnect all clients that are required to be */
 336         poll_disconnect_clients(msr);
 337
 338         Master_PollEvent(msr);
 339
 340         if (Neuro_LBufIsEmpty(msr->statuses))
 341         {
 342                 if (msr->callback)
 343                 {
 344                         if (msr->status->status == State_ClientDisconnect)
 345                         {
 346                                 TRACE("Sending the event to really disconnect the client");
 347                                 /* real disconnect event */
 348                                 /* Master_PushEvent(msr, msr->status->connection, 16); */
 349                                 add_client_disconnect(msr->status->connection);
 350                         }
 351                 }
 352
 353                 Status_Set(msr->status, State_NoData, NULL, 0, NULL);
 354         }
 355         else
 356         {
 357                 Status *buf;
 358
 359                 /* fetch the first element */
 360                 buf = Neuro_GiveLBuf(msr->statuses);
 361
 362                 Status_Move(buf, msr->status);
 363
 364                 Neuro_SCleanLBuf(msr->statuses, buf);
 365
 366                 if (msr->status->status == State_ClientDisconnect)
 367                 {
 368                         TRACE("Sending the event to really disconnect the client");
 369                         /* real disconnect event */
 370                         /* Master_PushEvent(msr, msr->status->connection, 16); */
 371                         add_client_disconnect(msr->status->connection);
 372                 }
 373                 else
 374                 {
 375                         TRACE(Neuro_s("Will send Status type %d -- data address %x", msr->status->status, msr->status->packet));
 376                 }
 377         }
 378
 379         /*
 380         if (Master_PollEvent(msr) > 0)
 381         {
 382                 NEURO_WARN("Master_PollEvent raised an error", NULL);
 383
 384                 Status_Set(msr->status, State_Disconnect, NULL, 0, NULL);
 385
 386                 return msr->status;
 387         }
 388         */
 389
 390         return msr->status;
 391 }
 392
 393 /*-------------------- Constructor Destructor ----------------------*/
 394
 395 /* see Master for the connection_type */
 396 Master *
 397 Master_Create(u32 connection_type)
 398 {
 399         Master *msr;
 400         Slave *slv;
 401
 402         msr = calloc(1, sizeof(Master));
 403
 404         slv = NULL;
 405
 406         msr->inclpacket_size = 0;
 407
 408         msr->protocolType = SOCK_STREAM;
 409
 410         msr->ep = Epoll_Create(sizeof(Slave));
 411
 412         /* NEURO_TRACE("Creating the connect events buffer", NULL); */
 413         Neuro_CreateEBuf(&msr->cevents);
 414
 415         if (connection_type >= TYPE_CLIENT)
 416                 connection_type = TYPE_CLIENT;
 417
 418         msr->type = connection_type;
 419
 420 #if WIN32
 421         {
 422                 int _err = 0;
 423
 424                 _err = WSAStartup(MAKEWORD(1, 1), &msr->wsaData);
 425
 426                 if (_err < 0)
 427                         return NULL;
 428         }
 429 #endif /* WIN32 */
 430
 431         msr->disco_clients = Neuro_CreateLBuf();
 432
 433         msr->statuses = Neuro_CreateLBuf();
 434         Neuro_SetcallbLBuf(msr->statuses, clean_statuses_elem);
 435
 436         msr->status = Status_Create();
 437
 438         msr->callback = NULL;
 439
 440         return msr;
 441 }
 442
 443 void
 444 Master_Destroy(Master *msr)
 445 {
 446         if (!msr)
 447                 return;
 448
 449 #ifdef WIN32
 450         WSACleanup();
 451 #endif /* WIN32 */
 452
 453
 454         Neuro_CleanEBuf(&msr->cevents);
 455
 456
 457         Neuro_CleanLBuf(msr->disco_clients);
 458
 459
 460         Neuro_CleanLBuf(msr->statuses);
 461
 462         Status_Destroy(msr->status);
 463
 464         if (msr->slave)
 465                 Slave_Destroy(msr->slave);
 466
 467         Epoll_Destroy(msr->ep);
 468
 469         free(msr);
 470 }