server/sensor.c

   1 /**
   2  * @file sensor.c
   3  * @purpose Implementation of sensor thread
   4  * TODO: Finalise implementation
   5  */
   6
   7
   8 #include "common.h"
   9 #include "sensor.h"
  10 #include "options.h"
  11 #include <math.h>
  12
  13 /** Array of sensors, initialised by Sensor_Init **/
  14 static Sensor g_sensors[NUMSENSORS]; //global to this file
  15 static const char * g_sensor_names[] = {"analog_test0","analog_test1","digital_test0"};
  16 /**
  17  * Read a data value from a sensor; block until value is read
  18  * @param sensor_id - The ID of the sensor
  19  * @param d - DataPoint to set
  20  * @returns NULL for digital sensors when data is unchanged, otherwise d
  21  */
  22 DataPoint * GetData(int sensor_id, DataPoint * d)
  23 {
  24         // switch based on the sensor_id at the moment for testing;
  25         // might be able to just directly access ADC from sensor_id?
  26         //TODO: Implement for real sensors
  27
  28
  29         //TODO: We should ensure the time is *never* allowed to change on the server if we use gettimeofday
  30         //              Another way people might think of getting the time is to count CPU cycles with clock()
  31         //              But this will not work because a) CPU clock speed may change on some devices (RPi?) and b) It counts cycles used by all threads
  32
  33         struct timeval t;
  34         gettimeofday(&t, NULL);
  35         d->time_stamp = (t.tv_sec - g_options.start_time.tv_sec) + 1e-6*(t.tv_usec - g_options.start_time.tv_usec);
  36
  37         // Make time relative
  38         //d->time_stamp.tv_sec -= g_options.start_time.tv_sec;
  39         //d->time_stamp.tv_usec -= g_options.start_time.tv_usec;
  40
  41         switch (sensor_id)
  42         {
  43                 case ANALOG_TEST0:
  44                 {
  45                         static int count = 0;
  46                         d->value = count++;
  47                         break;
  48                 }
  49                 case ANALOG_TEST1:
  50                         d->value = (double)(rand() % 100) / 100;
  51                         break;
  52
  53                 //TODO: For digital sensors, consider only updating when sensor is actually changed
  54                 case DIGITAL_TEST0:
  55                         d->value = t.tv_sec % 2;
  56                         break;
  57                 case DIGITAL_TEST1:
  58                         d->value = (t.tv_sec+1)%2;
  59                         break;
  60                 default:
  61                         Fatal("Unknown sensor id: %d", sensor_id);
  62                         break;
  63         }
  64         usleep(100000); // simulate delay in sensor polling
  65
  66         return d;
  67 }
  68
  69
  70 /**
  71  * Destroy a sensor
  72  * @param s - Sensor to destroy
  73  */
  74 void Destroy(Sensor * s)
  75 {
  76         // Maybe move the binary file into long term file storage?
  77         fclose(s->file);
  78 }
  79
  80
  81
  82 /**
  83  * Initialise a sensor
  84  * @param s - Sensor to initialise
  85  */
  86 void Init(Sensor * s, int id)
  87 {
  88         s->write_index = 0;
  89         s->id = id;
  90
  91         #define FILENAMESIZE 3
  92         char filename[FILENAMESIZE];
  93         if (s->id >= pow(10, FILENAMESIZE))
  94         {
  95                 Fatal("Too many sensors! FILENAMESIZE is %d; increase it and recompile.", FILENAMESIZE);
  96         }
  97
  98         pthread_mutex_init(&(s->mutex), NULL);
  99
 100         sprintf(filename, "%d", s->id);
 101         unlink(filename); //TODO: Move old files somewhere
 102
 103         s->file = fopen(filename, "a+b"); // open binary file
 104         Log(LOGDEBUG, "Initialised sensor %d; binary file is \"%s\"", id, filename);
 105 }
 106
 107
 108
 109
 110
 111 /**
 112  * Run the main sensor polling loop
 113  * @param arg - Cast to Sensor* - Sensor that the thread will handle
 114  * @returns NULL (void* required to use the function with pthreads)
 115  */
 116 void * Sensor_Main(void * arg)
 117 {
 118         Sensor * s = (Sensor*)(arg);
 119
 120         while (Thread_Runstate() == RUNNING) //TODO: Exit condition
 121         {
 122                 // The sensor will write data to a buffer until it is full
 123                 // Then it will open a file and dump the buffer to the end of it.
 124                 // Rinse and repeat
 125
 126                 // The reason I've added the buffer is because locks are expensive
 127                 // But maybe it's better to just write data straight to the file
 128                 // I'd like to do some tests by changing SENSOR_DATABUFSIZ
 129
 130                 while (s->write_index < SENSOR_DATABUFSIZ)
 131                 {
 132                         DataPoint * d = &(s->buffer[s->write_index]);
 133                         if (GetData(s->id, d) == NULL)
 134                         {
 135                                 Fatal("Error collecting data");
 136                         }
 137                         s->write_index += 1;
 138                 }
 139
 140                 //Log(LOGDEBUG, "Filled buffer");
 141
 142                 // CRITICAL SECTION (no threads should be able to read/write the file at the same time)
 143                 pthread_mutex_lock(&(s->mutex));
 144                         //TODO: Valgrind complains about this fseek: "Syscall param write(buf) points to uninitialised byte(s)"
 145                         //              Not sure why, but we should find out and fix it.
 146                         fseek(s->file, 0, SEEK_END);
 147                         int amount_written = fwrite(s->buffer, sizeof(DataPoint), SENSOR_DATABUFSIZ, s->file);
 148                         if (amount_written != SENSOR_DATABUFSIZ)
 149                         {
 150                                 Fatal("Wrote %d data points and expected to write %d to \"%s\" - %s", amount_written, SENSOR_DATABUFSIZ, strerror(errno));
 151                         }
 152                         //Log(LOGDEBUG, "Wrote %d data points for sensor %d", amount_written, s->id);
 153                 pthread_mutex_unlock(&(s->mutex));
 154                 // End of critical section
 155
 156                 s->write_index = 0; // reset position in buffer
 157
 158         }
 159         Log(LOGDEBUG, "Thread for sensor %d exits", s->id);
 160         return NULL;
 161 }
 162
 163 /**
 164  * Fill buffer with most recent sensor data
 165  * @param s - Sensor to use
 166  * @param buffer - Buffer to fill
 167  * @param bufsiz - Size of buffer to fill
 168  * @returns The number of DataPoints actually read
 169  */
 170 int Sensor_Query(Sensor * s, DataPoint * buffer, int bufsiz)
 171 {
 172         int amount_read = 0;
 173         //CRITICAL SECTION (Don't access file while sensor thread is writing to it!)
 174         pthread_mutex_lock(&(s->mutex));
 175
 176                 fseek(s->file, -bufsiz*sizeof(DataPoint), SEEK_END);
 177                 amount_read = fread(buffer, sizeof(DataPoint), bufsiz, s->file);
 178                 //Log(LOGDEBUG, "Read %d data points", amount_read);
 179         pthread_mutex_unlock(&(s->mutex));
 180         return amount_read;
 181 }
 182
 183 /**
 184  * Get a Sensor given an ID string
 185  * @param id_str ID string
 186  * @returns Sensor* identified by the string; NULL on error
 187  */
 188 Sensor * Sensor_Identify(const char * id_str)
 189 {
 190         char * end;
 191         // Parse string as integer
 192         int id = strtol(id_str, &end, 10);
 193         if (*end != '\0')
 194         {
 195                 return NULL;
 196         }
 197         // Bounds check
 198         if (id < 0 || id > NUMSENSORS)
 199                 return NULL;
 200
 201
 202         Log(LOGDEBUG, "Sensor \"%s\" identified", g_sensor_names[id]);
 203         return g_sensors+id;
 204 }
 205
 206 /**
 207  * Handle a request to the sensor module
 208  * @param context - The context to work in
 209  * @param params - Parameters passed
 210  */
 211 void Sensor_Handler(FCGIContext *context, char * params)
 212 {
 213         DataPoint buffer[SENSOR_QUERYBUFSIZ];
 214         StatusCodes status = STATUS_OK;
 215
 216         enum {DEFAULT, DUMP} operation = DEFAULT;
 217
 218         const char * key; const char * value;
 219
 220         Sensor * sensor = NULL;
 221
 222         while ((params = FCGI_KeyPair(params, &key, &value)) != NULL)
 223         {
 224                 Log(LOGDEBUG, "Got key=%s and value=%s", key, value);
 225                 if (strcmp(key, "id") == 0)
 226                 {
 227                         if (sensor != NULL)
 228                         {
 229                                 Log(LOGERR, "Only one sensor id should be specified");
 230                                 status = STATUS_ERROR;
 231                                 break;
 232                         }
 233                         if (*value == '\0')
 234                         {
 235                                 Log(LOGERR, "No id specified.");
 236                                 status = STATUS_ERROR;
 237                                 break;
 238                         }
 239
 240                         sensor = Sensor_Identify(value);
 241                         if (sensor == NULL)
 242                         {
 243                                 Log(LOGERR, "Invalid sensor id: %s", value);
 244                                 status = STATUS_ERROR;
 245                                 break;
 246                         }
 247                 }
 248                 else if (strcmp(key, "dump") == 0)
 249                 {
 250                         if (operation != DEFAULT)
 251                         {
 252                                 Log(LOGERR, "Operation already specified!");
 253                                 status = STATUS_ERROR;
 254                                 break;
 255                         }
 256                         operation = DUMP;
 257                 }
 258                 else
 259                 {
 260                         Log(LOGERR, "Unknown key \"%s\" (value = %s)", key, value);
 261                         status = STATUS_ERROR;
 262                         break;
 263                 }
 264         }
 265
 266         if (status != STATUS_ERROR && sensor == NULL)
 267         {
 268                 Log(LOGERR, "No valid sensor id given");
 269                 status = STATUS_ERROR;
 270         }
 271
 272         if (status == STATUS_ERROR)
 273         {
 274                 FCGI_RejectJSON(context);
 275                 return;
 276         }
 277
 278         switch (operation)
 279         {
 280                 case DUMP:
 281                 {
 282                         FCGI_PrintRaw("Content-type: text/plain\r\n\r\n");
 283                         //CRITICAL SECTION
 284                         pthread_mutex_lock(&(sensor->mutex));
 285                                 fseek(sensor->file, 0, SEEK_SET);
 286                                 int amount_read = 0;
 287                                 do
 288                                 {
 289                                         amount_read = fread(buffer, sizeof(DataPoint), SENSOR_QUERYBUFSIZ, sensor->file);
 290                                         for (int i = 0; i < amount_read; ++i)
 291                                         {
 292                                                 FCGI_PrintRaw("%f\t%f\n", buffer[i].time_stamp, buffer[i].value);
 293                                         }
 294
 295                                 }
 296                                 while (amount_read == SENSOR_QUERYBUFSIZ);
 297                         pthread_mutex_unlock(&(sensor->mutex));
 298                         // end critical section
 299                         break;
 300                 }
 301                 default:
 302                 {
 303                         FCGI_BeginJSON(context, status);
 304                         FCGI_JSONPair(key, value); // should spit back sensor ID
 305                         //Log(LOGDEBUG, "Call Sensor_Query...");
 306                         int amount_read = Sensor_Query(sensor, buffer, SENSOR_QUERYBUFSIZ);
 307                         //Log(LOGDEBUG, "Read %d DataPoints", amount_read);
 308                         //Log(LOGDEBUG, "Produce JSON response");
 309                         FCGI_JSONKey("data");
 310                         FCGI_JSONValue("[");
 311                         for (int i = 0; i < amount_read; ++i)
 312                         {
 313                                 //TODO: Consider; is it better to give both tv_sec and tv_usec to the client seperately, instead of combining here?
 314
 315                                 FCGI_JSONValue("[%f, %f]", buffer[i].time_stamp, buffer[i].value);
 316                                 if (i+1 < amount_read)
 317                                         FCGI_JSONValue(",");
 318                         }
 319                         FCGI_JSONValue("]");
 320                         //Log(LOGDEBUG, "Done producing JSON response");
 321                         FCGI_EndJSON();
 322                         break;
 323                 }
 324         }
 325 }
 326
 327 /**
 328  * Setup Sensors, start Sensor polling thread(s)
 329  */
 330 void Sensor_Spawn()
 331 {
 332         // start sensor threads
 333         for (int i = 0; i < NUMSENSORS; ++i)
 334         {
 335                 Init(g_sensors+i, i);
 336                 pthread_create(&(g_sensors[i].thread), NULL, Sensor_Main, (void*)(g_sensors+i));
 337         }
 338 }
 339
 340 /**
 341  * Quit Sensor loops
 342  */
 343 void Sensor_Join()
 344 {
 345         if (!Thread_Runstate())
 346         {
 347                 Fatal("This function should not be called before Thread_QuitProgram");
 348         }
 349         for (int i = 0; i < NUMSENSORS; ++i)
 350         {
 351                 pthread_join(g_sensors[i].thread, NULL);
 352                 Destroy(g_sensors+i);
 353         }
 354 }