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