src/ioman.c

   1 #include "include/ioman.h"
   2 #include <kernel.h>
   3 #include <string.h>
   4 #include <malloc.h>
   5 #include <stdio.h>
   6 #include <unistd.h>
   7 #ifdef __EESIO_DEBUG
   8 #include <sio.h>
   9 #endif
  10
  11 #define MAX_IO_REQUESTS 64
  12 #define MAX_IO_HANDLERS 64
  13
  14 extern void * _gp;
  15
  16 static int gIOTerminate = 0;
  17
  18 #define THREAD_STACK_SIZE   (8 * 1024)
  19
  20 static u8          disp_stack[THREAD_STACK_SIZE] ALIGNED(16);
  21 static u8          thread_stack[THREAD_STACK_SIZE] ALIGNED(16);
  22
  23 struct io_request_t {
  24         int  type;
  25         void *data;
  26         struct io_request_t *next;
  27 };
  28
  29 struct io_handler_t {
  30         int  type;
  31         io_request_handler_t handler;
  32 };
  33
  34 /// Circular request queue
  35 static struct io_request_t *gReqList;
  36 static struct io_request_t *gReqEnd;
  37
  38 static struct io_handler_t gRequestHandlers[MAX_IO_HANDLERS];
  39
  40
  41 static void ioThreadDispatcher ( void* apParam );
  42 static void ioAlarmFunc(s32 id, u16 time, void *arg);
  43
  44 static int gHandlerCount;
  45
  46 // id of the processing thread
  47 static s32 gIOThreadId;
  48 // id of the thread that handles thread switching
  49 static s32 gDispatcherThreadID;
  50
  51 // lock for tip processing
  52 static s32 gProcSemaId;
  53 // lock for queue end
  54 static s32 gEndSemaId;
  55 // ioPrintf sema id
  56 static s32 gIOPrintfSemaId;
  57
  58 static ee_thread_t gIOThread;
  59 static ee_sema_t gQueueSema;
  60
  61 static int isIOBlocked = 0;
  62 static int isIORunning = 0;
  63 static int alarmID = 0;
  64 static int stopIOTimer = 0;
  65
  66 int ioRegisterHandler(int type, io_request_handler_t handler) {
  67         WaitSema(gProcSemaId);
  68
  69         if (handler == NULL)
  70                 return IO_ERR_INVALID_HANDLER;
  71
  72         if (gHandlerCount >= MAX_IO_HANDLERS)
  73                 return IO_ERR_TOO_MANY_HANDLERS;
  74
  75         int i;
  76
  77         for (i = 0; i < gHandlerCount; ++i) {
  78                 if (gRequestHandlers[i].type == type)
  79                         return IO_ERR_DUPLICIT_HANDLER;
  80         }
  81
  82         gRequestHandlers[gHandlerCount].type = type;
  83         gRequestHandlers[gHandlerCount].handler = handler;
  84         gHandlerCount++;
  85
  86         SignalSema(gProcSemaId);
  87
  88         return IO_OK;
  89 }
  90
  91 static io_request_handler_t ioGetHandler(int type) {
  92         int i;
  93
  94         for (i = 0; i < gHandlerCount; ++i) {
  95                 struct io_handler_t* h = &gRequestHandlers[i];
  96
  97                 if (h->type == type)
  98                         return h->handler;
  99         }
 100
 101         return NULL;
 102 }
 103
 104 static void ioProcessRequest(struct io_request_t* req) {
 105         if (!req)
 106                 return;
 107
 108         io_request_handler_t hlr = ioGetHandler(req->type);
 109
 110         // invalidate the request
 111         void *data = req->data;
 112
 113         if (hlr)
 114                 hlr(data);
 115 }
 116
 117 static void ioWorkerThread(void) {
 118         while (!gIOTerminate) {
 119                 SleepThread();
 120
 121                 // no processing when io is blocked
 122                 if (isIOBlocked)
 123                         continue;
 124
 125                 // if term requested exit immediately from the loop
 126                 if (gIOTerminate)
 127                         break;
 128
 129                 // do we have a request in the queue?
 130                 while (gReqList) {
 131                         WaitSema(gProcSemaId);
 132
 133                         struct io_request_t* req = gReqList;
 134                         ioProcessRequest(req);
 135
 136                         // lock the queue tip as well now
 137                         WaitSema(gEndSemaId);
 138
 139                         // can't be sure if the request was
 140                         gReqList = req->next;
 141                         free(req);
 142
 143                         if (!gReqList)
 144                                 gReqEnd = NULL;
 145
 146                         SignalSema(gProcSemaId);
 147                         SignalSema(gEndSemaId);
 148                 }
 149         }
 150
 151         // delete the pending requests
 152         while (gReqList) {
 153                 struct io_request_t* req = gReqList;
 154                 gReqList = gReqList->next;
 155                 free(req); // TODO: Leak over here - we need a propper flag to free/not the user data
 156         }
 157
 158         // delete the semaphores
 159         DeleteSema(gProcSemaId);
 160         DeleteSema(gEndSemaId);
 161
 162         isIORunning = 0;
 163
 164         ExitDeleteThread();
 165 }
 166
 167 static void ioSimpleActionHandler(void *data) {
 168         io_simpleaction_t action = (io_simpleaction_t)data;
 169
 170         if (action)
 171                 action();
 172 }
 173
 174 void ioInit(void) {
 175         gIOTerminate = 0;
 176         gHandlerCount = 0;
 177         gReqList = NULL;
 178         gReqEnd = NULL;
 179
 180         gDispatcherThreadID = 0;
 181         gIOThreadId = 0;
 182
 183         gQueueSema.init_count = 1;
 184         gQueueSema.max_count = 1;
 185         gQueueSema.option = 0;
 186
 187         gProcSemaId = CreateSema(&gQueueSema);
 188         gEndSemaId = CreateSema(&gQueueSema);
 189         gIOPrintfSemaId = CreateSema(&gQueueSema);
 190
 191         ChangeThreadPriority (  GetThreadId (), 29 );
 192
 193         // default custom simple action handler
 194         ioRegisterHandler(IO_CUSTOM_SIMPLEACTION, &ioSimpleActionHandler);
 195
 196         memset(&gIOThread, 0, sizeof(gIOThread));
 197
 198         gIOThread.func         = (void *)ioThreadDispatcher;
 199         gIOThread.stack        = disp_stack;
 200         gIOThread.stack_size   = THREAD_STACK_SIZE;
 201         gIOThread.gp_reg       = &_gp;
 202         gIOThread.initial_priority = 0;
 203
 204         alarmID = SetAlarm( 625, &ioAlarmFunc, NULL);
 205
 206         // this one manages the thread switching in alarm intervals
 207         gDispatcherThreadID = CreateThread(&gIOThread);
 208         StartThread(gDispatcherThreadID, NULL);
 209
 210         gIOThread.func         = (void *)ioWorkerThread;
 211         gIOThread.stack        = thread_stack;
 212         gIOThread.initial_priority = 30;
 213
 214         isIORunning = 1;
 215         gIOThreadId = CreateThread(&gIOThread);
 216         StartThread(gIOThreadId, NULL);
 217
 218         ChangeThreadPriority (  GetThreadId (), 30 );
 219 }
 220
 221 int ioPutRequest(int type, void* data) {
 222         if (isIOBlocked)
 223                 return IO_ERR_IO_BLOCKED;
 224
 225         // check the type before queueing
 226         if (!ioGetHandler(type))
 227                 return IO_ERR_INVALID_HANDLER;
 228
 229         WaitSema(gEndSemaId);
 230
 231         // We don't have to lock the tip of the queue...
 232         // If it exists, it won't be touched, if it does not exist, it is not being processed
 233         struct io_request_t* req = gReqEnd;
 234
 235         if (!req) {
 236                 gReqList = (struct io_request_t*)malloc(sizeof(struct io_request_t));
 237                 req = gReqList;
 238                 gReqEnd = gReqList;
 239         } else {
 240                 req->next = (struct io_request_t*)malloc(sizeof(struct io_request_t));
 241                 req = req->next;
 242                 gReqEnd = req;
 243         }
 244
 245         req->next = NULL;
 246         req->type = type;
 247         req->data = data;
 248
 249         SignalSema(gEndSemaId);
 250
 251         WakeupThread(gIOThreadId);
 252         return IO_OK;
 253 }
 254
 255 int ioRemoveRequests(int type) {
 256         // TODO: This one needs free flag to stop leaks
 257         // lock the deletion sema and the queue end sema as well
 258         WaitSema(gEndSemaId);
 259         WaitSema(gProcSemaId);
 260
 261         int count = 0;
 262         struct io_request_t* req = gReqList;
 263         struct io_request_t* last = NULL;
 264
 265         while (req) {
 266                 if (req->type == type) {
 267                         struct io_request_t* next = req->next;
 268
 269                         if (last)
 270                                 last->next = next;
 271
 272                         if (req == gReqList)
 273                                 gReqList = next;
 274
 275                         if (req == gReqEnd)
 276                                 gReqEnd = last;
 277
 278                         count++;
 279                         free(req);
 280
 281                         req = next;
 282                 } else {
 283                         last = req;
 284                         req = req->next;
 285                 }
 286         }
 287
 288         SignalSema(gProcSemaId);
 289         SignalSema(gEndSemaId);
 290
 291         return count;
 292 }
 293
 294 void ioEnd(void) {
 295         // termination requested flag
 296         gIOTerminate = 1;
 297
 298         // wake up and wait for end
 299         WakeupThread(gIOThreadId);
 300         WakeupThread(gDispatcherThreadID);
 301
 302         // Cancel the thread dispatcher alarm (TODO: Is this correct?)
 303         ReleaseAlarm(alarmID);
 304 }
 305
 306 int ioGetPendingRequestCount(void) {
 307         int count = 0;
 308
 309         struct io_request_t* req = gReqList;
 310
 311         while (req) {
 312                 count++; req = req->next;
 313         }
 314
 315         return count;
 316 }
 317
 318 int ioHasPendingRequests(void) {
 319         return gReqList != NULL ? 1 : 0;
 320 }
 321
 322
 323 // From PS2DEV.ORG - Thread dispatcher
 324 static void ioThreadDispatcher ( void* apParam ) {
 325         while ( ! gIOTerminate ) {
 326                 SleepThread ();
 327         }  /* end while */
 328
 329         ExitDeleteThread();
 330 }  /* end dispatcher */
 331
 332 static void ioAlarmFunc(s32 id, u16 time, void *arg) {
 333         if (!gIOTerminate && !stopIOTimer) {
 334                 iWakeupThread ( gDispatcherThreadID );
 335                 iRotateThreadReadyQueue ( 30 );
 336                 alarmID = iSetAlarm ( 625, &ioAlarmFunc, NULL );
 337         }
 338 }
 339
 340 #ifdef __EESIO_DEBUG
 341 static char tbuf[2048];
 342 #endif
 343
 344 int ioPrintf(const char* format, ...) {
 345         WaitSema(gIOPrintfSemaId);
 346
 347         va_list args;
 348         va_start(args, format);
 349 #ifdef __EESIO_DEBUG
 350         int ret = vsnprintf((char *)tbuf, sizeof(tbuf), format, args);
 351         sio_putsn(tbuf);
 352 #else
 353         int ret = vprintf(format, args);
 354 #endif
 355         va_end(args);
 356
 357         SignalSema(gIOPrintfSemaId);
 358         return ret;
 359 }
 360
 361 int ioBlockOps(int block) {
 362         if (block && !isIOBlocked) {
 363                 isIOBlocked = 1;
 364
 365                 // wait for all io to finish
 366                 while (ioHasPendingRequests()) {
 367                         // TODO: This is in seconds, so can be too coarse
 368                         sleep(1);
 369                 }
 370
 371                 // now all io should be blocked
 372                 // stop the timer as well...
 373                 stopIOTimer = 1;
 374                 ReleaseAlarm(alarmID);
 375         } else if (!block && isIOBlocked) {
 376                 isIOBlocked = 0;
 377                 stopIOTimer = 0;
 378                 // create the alarm again
 379                 alarmID = SetAlarm( 625, &ioAlarmFunc, NULL);
 380         }
 381
 382
 383         return IO_OK;
 384 }