N-sim Emulation and simulation of Wireless Sensor Networks    Home    Project Page    Download    CVS    Installation    Configuration    Plug-ins  Hosted by

/home/brennan/n-sim/Vaike/linux/system-addons/system/control/sensor_ctl.c Go to the documentation of this file. /* * Copyright 2007. Los Alamos National Security, LLC. This material * was produced under U.S. Government contract DE-AC52-06NA25396 for * Los Alamos National Laboratory (LANL), which is operated by Los * Alamos National Security, LLC, for the Department of Energy. The * U.S. Government has rights to use, reproduce, and distribute this * software. NEITHER THE GOVERNMENT NOR LOS ALAMOS NATIONAL SECURITY, * LLC, MAKES ANY WARRANTY, EXPRESS OR IMPLIED, OR ASSUMES ANY LEGAL * LIABILITY FOR THE USE OF THIS SOFTWARE. If software is modified to * produce derivative works, such modified software should be clearly * marked, so as not to confuse it with the version available from LANL. * * Additionally, this program is free software; you can redistribute * it and/or modify it under the terms of the GNU General Public * License as published by the Free Software Foundation; version 2 of * the License. Accordingly, this program is distributed in the hope * it will be useful, but WITHOUT ANY WARRANTY; without even the * implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR * PURPOSE. See the GNU General Public License for more details. */ #include <stdio.h> #include <stdlib.h> #include <unistd.h> #include <string.h> #include <sys/time.h> #include <signal.h> #include <wait.h> #include <limits.h> #include <stdint.h> #include <gps.h> #include <pthread.h> #include <minisoap.h> #include <ipc.h> #include "sensor_ctl.h" #define BATTERY_DEV "/dev/platx/batmon" #define CONFIG_PATH "/etc/sensord.conf" #define GPS_PORT "2947" #define IPC_PORT_BASE 3133 struct { /* capabilities */ sensor_t *sensors; app_t *applications; /* gps data */ double fix_time; /* in seconds since epoch */ double satellites; double latitude; double longitude; double altitude; /* health data */ double voltage; int num_procs; app_t *processes; } status; static pthread_cond_t cmd_conditional; static pthread_mutex_t cmd_mutex; static SoapEvent *command; static pthread_cond_t data_conditional; static pthread_mutex_t data_mutex; static SoapEvent *data; static void launch_process(app_t *proc) { char cmdline[512]; sprintf(cmdline, "%s %u", proc->path, proc->ipc_port); if (!fork()) { system(cmdline); exit(1); } proc->running++; } static int query_gps(void) { struct gps_data_t *data = gps_open("127.0.0.1", GPS_PORT); if (data == NULL) return -1; gps_poll(data); status.fix_time = data->fix.time; status.satellites = data->satellites_used; if (data->fix.mode == MODE_2D || data->fix.mode == MODE_3D) { status.latitude = data->fix.latitude; status.longitude = data->fix.longitude; if (data->fix.mode == MODE_3D) status.altitude = data->fix.altitude; } gps_close(data); return 0; } static int query_battery(void) { char line[256]; FILE *batmon = fopen(BATTERY_DEV, "r"); if (fgets(line, 255, batmon) == NULL) return -1; if (sscanf(line, "%lfv", &status.voltage) < 1) return -1; fclose(batmon); return 0; } static int query_processes(void) { int i; for (i = 0; i < status.num_procs; i++) { int stat = 0; app_t *this_process = &status.processes[i]; char search[256]; sprintf(search, "ps ax | grep %s | grep -v grep >/dev/null", this_process->path); stat = system(search); if (WEXITSTATUS(stat) < 0) /* process not listed in results */ launch_process(this_process); } return 0; } static int status_response(char *server, int port) { char tmp[256]; SoapEvent evt; sensor_t *sensor; strcpy(evt.from, my_ip_addr); strcpy(evt.to, server); strcpy(evt.from_app, "sensord"); strcpy(evt.to_app, ""); evt.type = Event_Msg; sprintf(evt.msg, "<status>\n"); for (sensor = status.sensors; sensor != NULL; sensor = sensor->next) { if (strncmp(sensor->path+4, "gps", 3) == 0) { if (query_gps() == 0) { sprintf(tmp, "<gps>\n<latitude>%lf</latitude>\n" "<longitude>%lf</longitude>\n</gps>\n", status.latitude, status.longitude); strcat(evt.msg, tmp); } } if (strncmp(sensor->path+4, "battery", 7) == 0) { if (query_battery() == 0) { sprintf(tmp, "<battery>\n<volts>%lf</volts>\n" "</battery>\n", status.voltage); strcat(evt.msg, tmp); } } } if (query_processes() == 0) { app_t *proc; for (proc = status.processes; proc != NULL; proc = proc->next) { sprintf(tmp, "<process restarts=\"%d\">%s</process>\n", proc->running - 1, proc->path); strcat(evt.msg, tmp); } } strcat(evt.msg, "</status>"); return raise_soap_event(&evt, server, port); } static int provides_response(char *server, int port) { app_t *app; SoapEvent evt; strcpy(evt.from, my_ip_addr); strcpy(evt.to, "server"); strcpy(evt.from_app, "sensord"); strcpy(evt.to_app, ""); evt.type = Event_Msg; sprintf(evt.msg, "<provides>\n"); for (app = status.applications; app != NULL; app = app->next) { strcat(evt.msg, app->path); strncat(evt.msg, "\n", 1); } strcat(evt.msg, "</provides>"); return raise_soap_event(&evt, server, port); } static int data_register_handler(char *remainder) { app_t *proc; unsigned int port; char *appname = ""; char *sink_ip = ""; unsigned int found = 0; // FIXME: format is app:id or just app; sink_ip is evt.from if (sscanf(remainder, "%s %s:%d", appname, sink_ip, &port) < 3) { perror("sscanf"); return -1; } for (proc = status.processes; proc != NULL; proc = proc->next) { if (strcmp(appname, proc->path) == 0) { consumer_t *sink = (consumer_t*) malloc(sizeof(consumer_t)); strncpy(sink->host_ip, sink_ip, NET_ADDR_LEN); sink->port = port; sink->next = proc->sinks; proc->sinks = sink; found++; break; } } if (!found) { app_t *app; for (app = status.applications; app != NULL; app = app->next) { if (strcmp(appname, app->path) == 0) { consumer_t *sink; proc = (app_t*) malloc(sizeof(app_t)); memcpy(proc, app, sizeof(app_t)); proc->running = 0; proc->next = NULL; sink = (consumer_t*) malloc(sizeof(consumer_t)); strncpy(sink->host_ip, sink_ip, NET_ADDR_LEN); sink->port = port; sink->next = proc->sinks; proc->sinks = sink; proc->next = status.processes; status.processes = proc; found++; break; } } } if (!found) fprintf(stderr, "Attempt to register for %s - not found\n", appname); if (!proc->running) launch_process(proc); return 0; } #ifdef DEBUG static void print_config(FILE *out) { sensor_t *sensor; app_t *app; fprintf(out, "# sensord.conf\n"); fprintf(out, "\nipaddr:%s\n", my_ip_addr); for (sensor = status.sensors; sensor != NULL; sensor = sensor->next) { fprintf(out, "\nsensors:"); if (sensor->next != NULL) fprintf(out, "%s[%u],", sensor->path+4, sensor->ipc_port); else fprintf(out, "%s[%u];\n", sensor->path+4, sensor->ipc_port); } for (app = status.applications; app != NULL; app = app->next) { app_t *req; int paren = (app->required_apps != NULL ? 1 : 0); fprintf(out, "\nprovides:"); fprintf(out, "%s[%u]", app->path, app->ipc_port); if (paren) fprintf(out, "("); for (req = app->required_apps; req != NULL; req = req->next) { if (strcmp(req->host, "") == 0) fprintf(out, "%s[%u]", req->path, req->ipc_port); else fprintf(out, "%s:%s", req->path, req->host); } if (paren) fprintf(out, ")"); if(app->next != NULL) fprintf(out, ","); else fprintf(out, ";\n"); } } #endif /* soap message callback */ void sensor_ctl(SoapEvent *env) { if (strncmp(env->to, my_ip_addr, NET_ADDR_LEN) != 0) return; if (env->type == Event_Msg) { pthread_mutex_lock(&data_mutex); data = env; pthread_cond_signal(&data_conditional); pthread_mutex_unlock(&data_mutex); } else if (env->type == Command_Msg) { pthread_mutex_lock(&cmd_mutex); command = env; pthread_cond_signal(&cmd_conditional); pthread_mutex_unlock(&cmd_mutex); } usleep(100); /* for mutex settling */ } static int msg_route(SoapEvent *env) { if (strcmp(env->to, "") == 0) { // FIXME } else if (strcmp(env->to, "127.0.0.1") == 0 || strcmp(env->to, my_ip_addr) == 0) { /* local */ app_t *proc; for (proc = status.processes; proc != NULL; proc = proc->next) { if (strcmp(proc->path, env->to_app) == 0) { ipc_data_send(env); break; } } } else { /* remote */ raise_soap_event(env, env->to, DEFAULT_MINISOAP_PORT); } return 0; } static void *data_listener(void *unused) { while(!quitting()) { pthread_mutex_lock(&data_mutex); pthread_cond_wait(&data_conditional, &data_mutex); msg_route(data); free(data); pthread_mutex_unlock(&data_mutex); } return NULL; } static void *cmd_listener(void *unused) { while(!quitting()) { char *cmd = ""; char *remainder = ""; pthread_mutex_lock(&cmd_mutex); pthread_cond_wait(&cmd_conditional, &cmd_mutex); if (sscanf(command->msg, "%s %s", cmd, remainder) < 1) { perror("sscanf"); continue; } if (strncmp(cmd, "status", 6) == 0) status_response(command->from, DEFAULT_MINISOAP_PORT); else if (strncmp(cmd, "provides", 8) == 0) provides_response(command->from, DEFAULT_MINISOAP_PORT); else if (strncmp(cmd, "register", 8) == 0) data_register_handler(remainder); else msg_route(command); free(command); pthread_mutex_unlock(&cmd_mutex); } return NULL; } int main(int argc, char *argv[]) { int error = 0; char *config_name = CONFIG_PATH; FILE *config_file; int line_len = 255; char line[line_len + 1]; unsigned int ports_count = IPC_PORT_BASE; unsigned int under_sensors = 0; unsigned int under_provides = 0; unsigned int bad_config = 0; app_t *app; if (argc != 1) { fprintf(stderr, "Usage: %s\n", argv[0]); exit(-1); } status.sensors = NULL; status.applications = NULL; status.processes = NULL; pthread_mutex_init(&cmd_mutex, NULL); pthread_cond_init(&cmd_conditional, NULL); if ((error = initialize_ipc(-1)) < 0) return error; /* parse config file */ config_file = fopen(config_name, "r"); while (fgets(line, line_len, config_file) != NULL) { char *current = ""; char *remainder = ""; if (sscanf(line, "%*s:%s", remainder) < 1) { if (sscanf(line, "\t%s", remainder) < 1) continue; } if (strncmp(line, "#", 1) == 0) /* a comment line */ continue; else if (strncasecmp(line, "ipaddr", 6) == 0) sscanf(remainder, "%s;", my_ip_addr); else if (strncasecmp(line, "sensors", 7) == 0 || (strncmp(line, "\t", 1) == 0 && under_sensors)) { sensor_t *prev = status.sensors; while (sscanf(remainder, "%s, %s", current, remainder) > 0) { sensor_t *this_sensor = (sensor_t*) malloc(sizeof(sensor_t)); if (current[strlen(current)-1] == ';') { current[strlen(current)-1] = '\0'; under_sensors = 0; } else under_sensors++; strncpy(this_sensor->path, "raw-", 4); strncat(this_sensor->path, current, PATH_MAX-4); this_sensor->ipc_port = ports_count++; this_sensor->next = prev; prev = this_sensor; } status.sensors = prev; } else if (strncasecmp(line, "provides", 9) == 0 || (strncmp(line, "\t", 1) == 0 && under_provides)) { app_t *prev_app = status.applications; while (sscanf(remainder, "%s, %s", current, remainder) > 0) { app_t *prev_required = NULL; app_t *this_app = (app_t*) malloc(sizeof(app_t)); char *app = ""; char *support = ""; char *required = ""; if (current[strlen(current)-1] == ';') { current[strlen(current)-1] = '\0'; under_provides = 0; } else under_provides++; if (sscanf(current, "%s (%s)", app, support) < 2) app = current; strncpy(this_app->path, app, PATH_MAX); this_app->running = 0; memset(&this_app->host, '\0', NET_ADDR_LEN); this_app->ipc_port = ports_count++; this_app->sinks = NULL; this_app->next = prev_app; prev_app = this_app; while (sscanf(support, "%s, %s", required, support) > 0) { char *host_string = ""; app_t *a_new = (app_t*) malloc(sizeof(app_t)); if (sscanf(required, "%s:%s", required, host_string) > 1) strncpy(a_new->host, host_string, NET_ADDR_LEN); else a_new->ipc_port = -1; strncpy(a_new->path, required, PATH_MAX); a_new->running = 0; a_new->sinks = NULL; a_new->next = prev_required; prev_required = a_new; } } status.applications = prev_app; } } fclose(config_file); /* fill in ipc_port of required apps */ for (app = status.applications; app != NULL; app = app->next) { app_t *req_app; for (req_app = app->required_apps; req_app != NULL; req_app = req_app->next) { if (req_app->ipc_port < 0) { int found = 0; sensor_t *my_sens; app_t *my_apps; for (my_sens = status.sensors; my_sens != NULL; my_sens = my_sens->next) { if (strcmp(my_sens->path, req_app->path) == 0) { req_app->ipc_port = my_sens->ipc_port; found++; break; } } if (!found) { for (my_apps = status.applications; my_apps != NULL; my_apps = my_apps->next) { if (strcmp(my_apps->path, req_app->path) == 0) { req_app->ipc_port = my_apps->ipc_port; found++; break; } } } if (!found) { fprintf(stderr, "Error: %s requires " "%s which is missing from the " "configuration.\n", app->path, req_app->path); bad_config++; } } } } if (bad_config) exit(-1); #ifdef DEBUG print_config(stderr); #endif if (!fork()) { /* daemon mode */ int error = 0; pthread_t net_data_thread, net_cmd_thread; setsid(); if (register_soap_handler(sensor_ctl) < 0) { fprintf(stderr, "Soap server initialization failed\n"); exit(-1); } if (register_data_handler(IPC_DATA_PORT, msg_route) < 0) { fprintf(stderr, "Data server initialization failed\n"); exit(-1); } if (register_cmd_handler(IPC_CMD_PORT, msg_route) < 0) { fprintf(stderr, "Command server initialization failed\n"); exit(-1); } if ((error = pthread_create(&net_data_thread, NULL, data_listener, NULL)) < 0) { fprintf(stderr, "Data client initialization failed\n"); exit(-1); } if ((error = pthread_create(&net_cmd_thread, NULL, cmd_listener, NULL)) < 0) { fprintf(stderr, "Command client initialization failed\n"); exit(-1); } pthread_join(soap_thread, NULL); pthread_join(net_data_thread, NULL); pthread_join(net_cmd_thread, NULL); pthread_join(data_thread, NULL); pthread_join(cmd_thread, NULL); exit(0); } return 0; }

© 2007, Los Alamos National Security, LLC.