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

/home/brennan/n-sim/OrbisQuartus/control/mini_mpi_server.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 <errno.h> #include <netdb.h> #include <sys/types.h> #include <sys/socket.h> #include <sys/time.h> #include <netinet/in.h> #include <arpa/inet.h> #include <pthread.h> #include <mini_mpi.h> #include <daemon.h> #include "mini_mpi_server.h" #ifdef OQ_TESTING #define TIMEOUT 2 #else #define TIMEOUT 180 // in seconds #endif #define BACKLOG 10 struct timeval _beginning_; volatile unsigned int __first__; volatile unsigned int _proc_count_; pthread_cond_t mpi_conditional; pthread_mutex_t mpi_mutex; pthread_mutex_t task_mutex; typedef struct task_handle { pthread_t thread; int id; struct sockaddr_in ip_addr; int fd; short flag; struct task_handle *list; struct task_handle *next; } task_t; task_t *tasks_head, *tasks_tail; int get_cluster_size(void) { int size = 0; pthread_mutex_lock(&mpi_mutex); size = _proc_count_; pthread_mutex_lock(&mpi_mutex); return size; } int get_task_socket(int i) { task_t *t; int sock = -1; pthread_mutex_lock(&mpi_mutex); t = tasks_head; while (t != NULL) { if (t->id == i) { sock = t->fd; break; } t = t->next; } pthread_mutex_unlock(&mpi_mutex); return sock; } void change_task_flag(int i, short f) { task_t *t; pthread_mutex_lock(&mpi_mutex); t = tasks_head; while (t != NULL) { if (t->id == i) { t->flag = f; break; } t = t->next; } pthread_mutex_unlock(&mpi_mutex); } int tasks_to_string(int count, task_t *head, char *string) { int i, size = 0; int string_size = (sizeof(string) / sizeof(char)) - 1; task_t *t = head; for (i = 0; i < count; i++) { char tmp[65]; sprintf(tmp, "::%d:%s", t->id, inet_ntoa(t->ip_addr.sin_addr)); size += strlen(tmp); strncat(string, tmp, string_size); string_size -= strlen(tmp); t = t->next; } return size; } static void reset_timeout(void) { pthread_mutex_lock(&task_mutex); __first__ = 1; pthread_mutex_unlock(&task_mutex); } static int timedout(void) { int value = 0; pthread_mutex_lock(&task_mutex); if (__first__) { gettimeofday(&_beginning_, NULL); __first__ = 0; } else { struct timeval now; gettimeofday(&now, NULL); if (now.tv_sec - _beginning_.tv_sec >= TIMEOUT && now.tv_usec - _beginning_.tv_usec >= 0) value = 1; } pthread_mutex_unlock(&task_mutex); return value; } void *mpi_handler(void *t) { task_t *task = (task_t*)t; unsigned short done = 0; int task_strlen; pthread_mutex_lock(&mpi_mutex); // check for MINI_MPI_INIT request // // wait for timeout pthread_cond_signal(&mpi_conditional); pthread_mutex_unlock(&mpi_mutex); task_strlen = tasks_to_string(_proc_count_, tasks_head, NULL); char msg[task_strlen + 128]; char tasks[task_strlen + 1]; tasks_to_string(_proc_count_, tasks_head, tasks); sprintf(msg, "%d::%d:%d::%s", MINI_MPI_HEAD, _proc_count_, task->id, tasks); send(task->fd, msg, strlen(msg), 0); while (!done > 0 && !daemon_shuttingdown()) { int req_id = 0; // recv requests: switch (req_id) { case MINI_MPI_INIT: pthread_mutex_lock(&task_mutex); // log error pthread_mutex_unlock(&task_mutex); break; case MINI_MPI_ABORT: // broadcast MINI_MPI_ABORT break; case MINI_MPI_BARRIER: pthread_mutex_lock(&task_mutex); // wait for all pthread_mutex_unlock(&task_mutex); // broadcast MINI_MPI_RESUME break; case MINI_MPI_SEND: case MINI_MPI_RECV: case MINI_MPI_SCATTER: case MINI_MPI_GATHER: pthread_mutex_lock(&task_mutex); // log msg to file pthread_mutex_unlock(&task_mutex); break; case MINI_MPI_FINAL: done = 1; break; default: pthread_mutex_lock(&task_mutex); // log error pthread_mutex_unlock(&task_mutex); break; } } close(task->fd); return NULL; } void *mini_mpi_server(void *arg_struct) { int argc = 0; char **argv; if (arg_struct != NULL) { argc = ((struct arguments *)arg_struct)->argc; argv = ((struct arguments *)arg_struct)->argv; } pthread_mutex_init(&mpi_mutex, NULL); pthread_mutex_init(&task_mutex, NULL); while (!daemon_shuttingdown()) { task_t *h; int rc, listen_sd, handle_sd, yes = 1; struct sockaddr_in my_addr, their_addr; socklen_t sin_size = sizeof(struct sockaddr_in); tasks_head = tasks_tail = NULL; pthread_mutex_lock(&mpi_mutex); _proc_count_ = 0; if ((listen_sd = socket(PF_INET, SOCK_STREAM, 0)) == -1) { perror("socket"); exit(-1); } if (setsockopt(listen_sd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)) == -1) { perror("setsockopt"); exit(-1); } my_addr.sin_family = AF_INET; my_addr.sin_port = htons(MPI_PORT); my_addr.sin_addr.s_addr = INADDR_ANY; memset(&(my_addr.sin_zero), '\0', 8); if (bind(listen_sd, (struct sockaddr*) &my_addr, sizeof(struct sockaddr)) == -1) { perror("bind"); exit(-1); } if (listen(listen_sd, BACKLOG) == -1) { perror("listen"); exit(-1); } while (!timedout() || !daemon_quitting() || !daemon_shuttingdown()) { task_t *tmp; if ((handle_sd = accept(listen_sd, (struct sockaddr*) &their_addr, &sin_size)) == -1) { perror("accept"); exit(-1); } tmp = tasks_tail; tasks_tail = (task_t*)malloc(sizeof(task_t)); pthread_mutex_lock(&task_mutex); tasks_tail->id = _proc_count_++; pthread_mutex_unlock(&task_mutex); tasks_tail->ip_addr = their_addr; tasks_tail->flag = 0; tasks_tail->fd = handle_sd; tasks_tail->next = NULL; if (tmp != NULL) tmp->next = tasks_tail; if (tasks_head == NULL) tasks_head = tasks_tail; tasks_tail->list = tasks_head; if ((rc = pthread_create(&tasks_tail->thread, NULL, mpi_handler, tasks_tail)) != 0) { fprintf(stderr, "ERROR-- MPI handler thread: %d\n", rc); exit(-1); } } pthread_mutex_unlock(&mpi_mutex); h = tasks_head; while (h != NULL) { task_t *tmp = h->next; pthread_join(h->thread, NULL); free(h); h = tmp; } close(listen_sd); } return NULL; } #ifdef OQ_TESTING #include <unistd.h> int mpi_test() { int error = 0; int flag = 0; MPI_Init(0, NULL); MPI_Initialized(&flag); if (flag == 0) { fprintf(stderr, "MPI initialization failed\n"); error = -1; } return error; } int main(int argc, char *argv[]) { int error, rc = 0; pthread_t mpi; struct arguments args; args.argc = argc; args.argv = argv; daemon_init(); reset_timeout(); if ((rc = pthread_create(&mpi, NULL, mini_mpi_server, NULL)) != 0) { fprintf(stderr, "ERROR-- mini MPI server thread: %d\n", rc); exit(-1); } error = mpi_test(); daemon_shutdown(); pthread_join(mpi, NULL); return error; } #endif

© 2007, Los Alamos National Security, LLC.