From da4d0280e81a4966c24ad3fc602d00f97739473a Mon Sep 17 00:00:00 2001 From: nvt Date: Tue, 6 Dec 2011 20:27:41 +0100 Subject: [PATCH] NetDB: a simple DBMS application using Rime's Mesh protocol to transfer queries and responses. --- examples/antelope/netdb/Makefile | 6 + examples/antelope/netdb/netdb-client.c | 133 +++++++++++ examples/antelope/netdb/netdb-server.c | 300 +++++++++++++++++++++++++ examples/antelope/netdb/netdb.csc | 271 ++++++++++++++++++++++ examples/antelope/netdb/project-conf.h | 20 ++ 5 files changed, 730 insertions(+) create mode 100644 examples/antelope/netdb/Makefile create mode 100644 examples/antelope/netdb/netdb-client.c create mode 100644 examples/antelope/netdb/netdb-server.c create mode 100755 examples/antelope/netdb/netdb.csc create mode 100644 examples/antelope/netdb/project-conf.h diff --git a/examples/antelope/netdb/Makefile b/examples/antelope/netdb/Makefile new file mode 100644 index 000000000..e680009ca --- /dev/null +++ b/examples/antelope/netdb/Makefile @@ -0,0 +1,6 @@ +CONTIKI = ../../../ +APPS += antelope +CFLAGS += -Wall -g -DPROJECT_CONF_H=\"project-conf.h\" +SMALL = 1 + +include $(CONTIKI)/Makefile.include diff --git a/examples/antelope/netdb/netdb-client.c b/examples/antelope/netdb/netdb-client.c new file mode 100644 index 000000000..59fdf9277 --- /dev/null +++ b/examples/antelope/netdb/netdb-client.c @@ -0,0 +1,133 @@ +/* + * Copyright (c) 2011, Swedish Institute of Computer Science + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions + * are met: + * 1. Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * 3. Neither the name of the Institute nor the names of its contributors + * may be used to endorse or promote products derived from this software + * without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE INSTITUTE AND CONTRIBUTORS ``AS IS'' AND + * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE INSTITUTE OR CONTRIBUTORS BE LIABLE + * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL + * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS + * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) + * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT + * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY + * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF + * SUCH DAMAGE. + */ + +/** + * \file + * A small command-line interface for the querying remote database systems. + * \author + * Nicolas Tsiftes + */ + +#include + +#include "contiki.h" +#include "dev/serial-line.h" +#include "net/rime.h" +#include "net/rime/mesh.h" + +#include "antelope.h" +/*---------------------------------------------------------------------------*/ +#define MAX_QUERY_SIZE 100 + +#define NETDB_CHANNEL 70 + +#ifndef SERVER_ID +#define SERVER_ID 4 +#endif +/*---------------------------------------------------------------------------*/ +PROCESS(netdb_process, "NetDB"); +AUTOSTART_PROCESSES(&netdb_process); + +static unsigned server_id = SERVER_ID; +static struct mesh_conn mesh; +/*---------------------------------------------------------------------------*/ +PROCESS(shell_process, "Shell Process"); + +PROCESS_THREAD(shell_process, ev, data) +{ + rimeaddr_t addr; + + PROCESS_BEGIN(); + + printf("NetDB client\n"); + + for(;;) { + PROCESS_WAIT_EVENT_UNTIL(ev == serial_line_event_message && data != NULL); + if(strncmp(data, "server ", 7) == 0) { + server_id = atoi((char *)data + 7); + } else { + printf("%lu Transmitting query \"%s\" to node %u\n", clock_time(), (char *)data, server_id); + packetbuf_copyfrom(data, strlen(data)); + addr.u8[0] = server_id; + addr.u8[1] = 0; + packetbuf_set_attr(PACKETBUF_ATTR_PACKET_TYPE, + PACKETBUF_ATTR_PACKET_TYPE_STREAM); + mesh_send(&mesh, &addr); + } + } + + PROCESS_END(); +} + +/*---------------------------------------------------------------------------*/ +static void +sent(struct mesh_conn *c) +{ +} + +static void +timedout(struct mesh_conn *c) +{ + printf("Failed to send packet: time out\n"); +} + +static void +received(struct mesh_conn *c, const rimeaddr_t *from, uint8_t hops) +{ + char *data; + unsigned len; + static char reply[MAX_QUERY_SIZE + 1]; + + data = (char *)packetbuf_dataptr(); + len = packetbuf_datalen(); + + if(len > MAX_QUERY_SIZE) { + printf("Too long query: %d bytes\n", len); + return; + } + + memcpy(reply, data, len); + reply[len] = '\0'; + + printf("%lu Reply received from %d.%d (%d hops): %s", + clock_time(), from->u8[0], from->u8[1], (int)hops, reply); +} + +static const struct mesh_callbacks callbacks = {received, sent, timedout}; +/*---------------------------------------------------------------------------*/ +PROCESS_THREAD(netdb_process, ev, data) +{ + PROCESS_EXITHANDLER(mesh_close(&mesh)); + PROCESS_BEGIN(); + + mesh_open(&mesh, NETDB_CHANNEL, &callbacks); + process_start(&shell_process, NULL); + + PROCESS_END(); +} diff --git a/examples/antelope/netdb/netdb-server.c b/examples/antelope/netdb/netdb-server.c new file mode 100644 index 000000000..50700a457 --- /dev/null +++ b/examples/antelope/netdb/netdb-server.c @@ -0,0 +1,300 @@ +/* + * Copyright (c) 2011, Swedish Institute of Computer Science + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions + * are met: + * 1. Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * 3. Neither the name of the Institute nor the names of its contributors + * may be used to endorse or promote products derived from this software + * without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE INSTITUTE AND CONTRIBUTORS ``AS IS'' AND + * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE INSTITUTE OR CONTRIBUTORS BE LIABLE + * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL + * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS + * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) + * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT + * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY + * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF + * SUCH DAMAGE. + */ + +/** + * \file + * A small command-line interface for the querying remote database systems. + * \author + * Nicolas Tsiftes + */ + +#include + +#include "contiki.h" +#include "dev/serial-line.h" +#include "dev/sht11.h" +#include "lib/random.h" +#include "net/rime.h" +#include "net/rime/mesh.h" + +#include "antelope.h" +/*---------------------------------------------------------------------------*/ +/* Sampling interval in Hz. */ +#ifndef SAMPLING_INTERVAL +#define SAMPLING_INTERVAL 60 +#endif + +#ifndef RESPONSE_LIMIT +#define RESPONSE_LIMIT 1000 +#endif + +#ifndef PREPARE_DB +#define PREPARE_DB 1 +#endif + +#ifndef CARDINALITY +#define CARDINALITY 1000 +#endif + +#define MAX_BUFFER_SIZE 80 + +#define NETDB_CHANNEL 70 +/*---------------------------------------------------------------------------*/ +PROCESS(netdb_process, "NetDB"); +AUTOSTART_PROCESSES(&netdb_process); + +static struct mesh_conn mesh; +static rimeaddr_t reply_addr; +static uint8_t buffer_offset; +static char buffer[MAX_BUFFER_SIZE]; +/*---------------------------------------------------------------------------*/ +static void +send_buffered_data(void) +{ + if(buffer_offset > 0) { + packetbuf_copyfrom(buffer, buffer_offset); + mesh_send(&mesh, &reply_addr); + buffer_offset = 0; + } +} +/*---------------------------------------------------------------------------*/ +static int +buffer_db_data(const char *format, ...) +{ + va_list ap; + size_t len; + char tmp[MAX_BUFFER_SIZE + 1]; + + va_start(ap, format); + len = vsnprintf(tmp, sizeof(tmp), format, ap); + va_end(ap); + + if(len < 0) { + return -1; + } + + if(len + buffer_offset > sizeof(buffer)) { + send_buffered_data(); + } + + memcpy(&buffer[buffer_offset], tmp, len); + buffer_offset += len; + + return len; +} +/*---------------------------------------------------------------------------*/ +static void +take_sample(void) +{ + unsigned seconds; + unsigned humidity; + + seconds = clock_seconds(); + humidity = /*sht11_humidity()*/ random_rand(); + if(DB_ERROR(db_query(NULL, "INSERT (%u, %u) INTO samples;", + seconds, humidity))) { + printf("DB insertion failed\n"); + } +} +/*---------------------------------------------------------------------------*/ +static void +stop_handler(void *ptr) +{ + printf("END\n"); +} +/*---------------------------------------------------------------------------*/ +PROCESS(query_process, "Query process"); + +PROCESS_THREAD(query_process, ev, data) +{ + static db_handle_t handle; + db_result_t result; + static tuple_id_t matching; + static tuple_id_t processed; +#if !PREPARE_DB + static struct etimer sampling_timer; +#endif + static unsigned i, errors; + + PROCESS_BEGIN(); + + printf("NetDB host\n"); + + db_init(); + db_set_output_function(buffer_db_data); + + db_query(NULL, "REMOVE RELATION samples;"); + db_query(NULL, "CREATE RELATION samples;"); + db_query(NULL, "CREATE ATTRIBUTE time DOMAIN INT IN samples;"); + db_query(NULL, "CREATE ATTRIBUTE hum DOMAIN INT IN samples;"); + db_query(NULL, "CREATE INDEX samples.time TYPE INLINE;"); + +#if PREPARE_DB + printf("Preparing the DB with %d tuples...\n", CARDINALITY); + errors = 0; + for(i = 1; i <= CARDINALITY; i++) { + PROCESS_PAUSE(); + + result = db_query(NULL, "INSERT (%u, %u) INTO samples;", + i, (unsigned)random_rand()); + if(DB_ERROR(result)) { + errors++; + } + } + printf("Done. Insertion errors: %d\n", errors); + printf("Ready to process queries\n"); +#else + etimer_set(&sampling_timer, SAMPLING_INTERVAL * CLOCK_SECOND); +#endif + + for(;;) { + PROCESS_WAIT_EVENT(); + + if(ev == serial_line_event_message && data != NULL) { + printf("START %s\n", (char *)data); + result = db_query(&handle, data); + if(DB_ERROR(result)) { + buffer_db_data("Query error: %s\n", db_get_result_message(result)); + stop_handler(NULL); + db_free(&handle); + continue; + } + + if(!db_processing(&handle)) { + buffer_db_data("OK\n"); + send_buffered_data(); + stop_handler(NULL); + continue; + } + + packetbuf_set_attr(PACKETBUF_ATTR_PACKET_TYPE, + PACKETBUF_ATTR_PACKET_TYPE_STREAM); + + db_print_header(&handle); + + matching = 0; + processed = 0; + + while(db_processing(&handle)) { + PROCESS_PAUSE(); + + if(matching == RESPONSE_LIMIT) { + buffer_db_data("Response suppressed at %u tuples: limit reached\n", + RESPONSE_LIMIT); + stop_handler(NULL); + db_free(&handle); + break; + } + + result = db_process(&handle); + if(result == DB_GOT_ROW) { + /* The processed tuple matched the condition in the query. */ + matching++; + processed++; + db_print_tuple(&handle); + } else if(result == DB_OK) { + /* A tuple was processed, but did not match the condition. */ + processed++; + continue; + } else { + if(result == DB_FINISHED) { + /* The processing has finished. Wait for a new command. */ + buffer_db_data("[%ld tuples returned; %ld tuples processed]\n", + (long)matching, (long)processed); + buffer_db_data("OK\n"); + } else if(DB_ERROR(result)) { + buffer_db_data("Processing error: %s\n", + db_get_result_message(result)); + } + stop_handler(NULL); + db_free(&handle); + } + } + send_buffered_data(); + } + +#if !PREPARE_DB + if(etimer_expired(&sampling_timer)) { + take_sample(); + etimer_reset(&sampling_timer); + } +#endif + } + + PROCESS_END(); +} +/*---------------------------------------------------------------------------*/ +static void +sent(struct mesh_conn *c) +{ +} + +static void +timedout(struct mesh_conn *c) +{ + printf("packet time out\n"); +} + +static void +received(struct mesh_conn *c, const rimeaddr_t *from, uint8_t hops) +{ + char *data; + unsigned len; + static char query[MAX_BUFFER_SIZE + 1]; + + data = (char *)packetbuf_dataptr(); + len = packetbuf_datalen(); + + if(len > MAX_BUFFER_SIZE) { + buffer_db_data("Too long query: %d bytes\n", len); + return; + } + + memcpy(query, data, len); + query[len] = '\0'; + + printf("Query received from %d.%d: %s (%d hops)\n", + from->u8[0], from->u8[1], query, (int)hops); + rimeaddr_copy(&reply_addr, from); + + process_post(&query_process, serial_line_event_message, query); +} + +static const struct mesh_callbacks callbacks = {received, sent, timedout}; +/*---------------------------------------------------------------------------*/ +PROCESS_THREAD(netdb_process, ev, data) +{ + PROCESS_EXITHANDLER(mesh_close(&mesh)); + PROCESS_BEGIN(); + + mesh_open(&mesh, NETDB_CHANNEL, &callbacks); + process_start(&query_process, NULL); + + PROCESS_END(); +} diff --git a/examples/antelope/netdb/netdb.csc b/examples/antelope/netdb/netdb.csc new file mode 100755 index 000000000..d6d98a0dd --- /dev/null +++ b/examples/antelope/netdb/netdb.csc @@ -0,0 +1,271 @@ + + + [CONTIKI_DIR]/tools/cooja/apps/mrm + [CONTIKI_DIR]/tools/cooja/apps/mspsim + [CONTIKI_DIR]/tools/cooja/apps/avrora + [CONTIKI_DIR]/tools/cooja/apps/serial_socket + [CONTIKI_DIR]/tools/cooja/apps/collect-view + + NetDB + 0 + 123456 + 1000000 + + se.sics.cooja.radiomediums.UDGM + 30.0 + 30.0 + 1.0 + 1.0 + + + 40000 + + + se.sics.cooja.mspmote.SkyMoteType + sky1 + NetDB Server + [CONTIKI_DIR]/examples/antelope/netdb/netdb-server.c + make netdb-server.sky TARGET=sky + [CONTIKI_DIR]/examples/antelope/netdb/netdb-server.sky + se.sics.cooja.interfaces.Position + se.sics.cooja.interfaces.RimeAddress + se.sics.cooja.interfaces.IPAddress + se.sics.cooja.interfaces.Mote2MoteRelations + se.sics.cooja.interfaces.MoteAttributes + se.sics.cooja.mspmote.interfaces.MspClock + se.sics.cooja.mspmote.interfaces.MspMoteID + se.sics.cooja.mspmote.interfaces.SkyButton + se.sics.cooja.mspmote.interfaces.SkyFlash + se.sics.cooja.mspmote.interfaces.SkyCoffeeFilesystem + se.sics.cooja.mspmote.interfaces.SkyByteRadio + se.sics.cooja.mspmote.interfaces.MspSerial + se.sics.cooja.mspmote.interfaces.SkyLED + se.sics.cooja.mspmote.interfaces.MspDebugOutput + se.sics.cooja.mspmote.interfaces.SkyTemperature + + + se.sics.cooja.mspmote.SkyMoteType + sky2 + NetDB Client + [CONTIKI_DIR]/examples/antelope/netdb/netdb-client.c + make netdb-client.sky TARGET=sky + [CONTIKI_DIR]/examples/antelope/netdb/netdb-client.sky + se.sics.cooja.interfaces.Position + se.sics.cooja.interfaces.RimeAddress + se.sics.cooja.interfaces.IPAddress + se.sics.cooja.interfaces.Mote2MoteRelations + se.sics.cooja.interfaces.MoteAttributes + se.sics.cooja.mspmote.interfaces.MspClock + se.sics.cooja.mspmote.interfaces.MspMoteID + se.sics.cooja.mspmote.interfaces.SkyButton + se.sics.cooja.mspmote.interfaces.SkyFlash + se.sics.cooja.mspmote.interfaces.SkyCoffeeFilesystem + se.sics.cooja.mspmote.interfaces.SkyByteRadio + se.sics.cooja.mspmote.interfaces.MspSerial + se.sics.cooja.mspmote.interfaces.SkyLED + se.sics.cooja.mspmote.interfaces.MspDebugOutput + se.sics.cooja.mspmote.interfaces.SkyTemperature + + + + + se.sics.cooja.interfaces.Position + 23.57340748739308 + 46.80222047486912 + 0.0 + + + se.sics.cooja.mspmote.interfaces.MspMoteID + 1 + + sky1 + + + + + se.sics.cooja.interfaces.Position + 40.39130096157144 + 70.54634688655467 + 0.0 + + + se.sics.cooja.mspmote.interfaces.MspMoteID + 2 + + sky1 + + + + + se.sics.cooja.interfaces.Position + 66.04131381969006 + 36.41113701058369 + 0.0 + + + se.sics.cooja.mspmote.interfaces.MspMoteID + 3 + + sky1 + + + + + se.sics.cooja.interfaces.Position + 63.00130046120498 + 80.89331313174746 + 0.0 + + + se.sics.cooja.mspmote.interfaces.MspMoteID + 4 + + sky1 + + + + + se.sics.cooja.interfaces.Position + 40.2894982777653 + 95.14334789567525 + 0.0 + + + se.sics.cooja.mspmote.interfaces.MspMoteID + 5 + + sky1 + + + + + se.sics.cooja.interfaces.Position + -13.168104050312492 + 40.250683112542255 + 0.0 + + + se.sics.cooja.mspmote.interfaces.MspMoteID + 6 + + sky1 + + + + + se.sics.cooja.interfaces.Position + 80.95025965975177 + 44.99507552455861 + 0.0 + + + se.sics.cooja.mspmote.interfaces.MspMoteID + 7 + + sky1 + + + + + se.sics.cooja.interfaces.Position + 6.857316697020866 + 33.24863334754029 + 0.0 + + + se.sics.cooja.mspmote.interfaces.MspMoteID + 8 + + sky1 + + + + + se.sics.cooja.interfaces.Position + 35.975659895989395 + 27.42171932830696 + 0.0 + + + se.sics.cooja.mspmote.interfaces.MspMoteID + 9 + + sky1 + + + + + se.sics.cooja.interfaces.Position + 13.672853648109518 + 68.2461872644317 + 0.0 + + + se.sics.cooja.mspmote.interfaces.MspMoteID + 10 + + sky1 + + + + + se.sics.cooja.interfaces.Position + 44.62423029692567 + 48.53691502749644 + 0.0 + + + se.sics.cooja.mspmote.interfaces.MspMoteID + 51 + + sky2 + + + + se.sics.cooja.plugins.SimControl + 259 + 3 + 205 + 0 + 0 + + + se.sics.cooja.plugins.Visualizer + + se.sics.cooja.plugins.skins.IDVisualizerSkin + se.sics.cooja.plugins.skins.MoteTypeVisualizerSkin + se.sics.cooja.plugins.skins.UDGMVisualizerSkin + 4.472125038273293 0.0 0.0 4.472125038273293 79.43486237544504 -89.06315297501011 + + 475 + 0 + 429 + 644 + 9 + + + se.sics.cooja.plugins.LogListener + + ID:4$ + + 1024 + 2 + 150 + 0 + 389 + + + se.sics.cooja.plugins.MoteInterfaceViewer + 10 + + Serial port + 0,0 + + 588 + 1 + 399 + 505 + 520 + + + diff --git a/examples/antelope/netdb/project-conf.h b/examples/antelope/netdb/project-conf.h new file mode 100644 index 000000000..9d1ed0c56 --- /dev/null +++ b/examples/antelope/netdb/project-conf.h @@ -0,0 +1,20 @@ +#undef QUEUEBUF_CONF_NUM +#define QUEUEBUF_CONF_NUM 4 + +#undef NETSTACK_CONF_RDC +#define NETSTACK_CONF_RDC nullrdc_driver + +#undef NETSTACK_CONF_RDC_CHANNEL_CHECK_RATE +#define NETSTACK_CONF_RDC_CHANNEL_CHECK_RATE 4 + +#undef DCOSYNC_CONF_ENABLED +#define DCOSYNC_CONF_ENABLED 0 + +#undef DB_FEATURE_JOIN +#define DB_FEATURE_JOIN 0 + +#undef RF_CHANNEL +#define RF_CHANNEL 16 + +#undef ROUTE_CONF_DEFAULT_LIFETIME +#define ROUTE_CONF_DEFAULT_LIFETIME 300