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/server/l4/mini_mpi-l4.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 <string.h> #include <errno.h> #include <mini_mpi.h> #include <mini_mpi-internal.h> #include "lwip.h" extern int my_numprocs; extern int my_taskid; extern unsigned short my_init; extern struct sockaddr_in head_addr; extern socket_ptr_t my_socket; extern task_t *task_array; //#define VIRT_TIME //#define MINI_MPI_TRACK #define MY_NC ((struct netconn*)my_socket.nc) static struct netconn *bcast_conn; int MPI_Init(int *argc, char ***argv) { int error = MPI_SUCCESS; int header, their_port = 0; struct netbuf *outbuf, *inbuf; struct ip_addr bcast_addr, *their_addr; char *table, msg_out[32], msg_in[MAX_MTU+1]; sprintf(msg_out, "%d", MINI_MPI_INIT); if ((outbuf = netbuf_new()) == NULL) { error = ENOMEM; goto end; } if ((bcast_conn = netconn_new(NETCONN_UDP)) == NULL) { error = ENOMEM; goto end_init; } IP4_ADDR(&bcast_addr, 255,255,255,255); if ((error = netconn_connect(bcast_conn, &bcast_addr, MPI_INIT_PORT)) < 0) { netconn_delete(bcast_conn); goto end_init; } netbuf_ref(outbuf, msg_out, strlen(msg_out) + 1); if ((error = netconn_send(bcast_conn, outbuf)) < 0) { netconn_close(bcast_conn); netconn_delete(bcast_conn); goto end_init; } await_reply: while ((inbuf = netconn_recv(bcast_conn)) != NULL) { int msg_len = sizeof(msg_in) - 1; int cur_len = 0; char indata[msg_len + 1]; do { char *fragdata; unsigned short len = 0; int left = msg_len - cur_len; netbuf_data(inbuf, (void**)&fragdata, &len); memcpy(msg_in + cur_len, fragdata, len > left ? left : len); cur_len += len; } while(netbuf_next(inbuf) >= 0); strncat(msg_in, indata, cur_len); if (their_port == 0) { their_addr = netbuf_fromaddr(inbuf); memcpy(&head_addr.sin_addr, their_addr, sizeof(struct ip_addr)); head_addr.sin_port = netbuf_fromport(inbuf); } free(inbuf); } error = sscanf(msg_in, "%d::%d:%d::", &header, &my_numprocs, &my_taskid); if (error < 3 || header != MINI_MPI_HEAD) { // log error goto await_reply; } if ((table = strstr(msg_in, "::::")) == NULL) { error = errno; netconn_close(bcast_conn); netconn_delete(bcast_conn); goto end_init; } table += 2; string_to_tasks(my_numprocs, table, task_array); if ((my_socket.nc = netconn_new(NETCONN_TCP)) == NULL) { error = ENOMEM; netconn_close(bcast_conn); netconn_delete(bcast_conn); goto end_init; } if ((error = netconn_connect(MY_NC, (struct ip_addr*)&head_addr.sin_addr, MPI_PORT)) < 0) { netconn_delete(MY_NC); netconn_close(bcast_conn); netconn_delete(bcast_conn); goto end_init; } #ifdef VIRT_TIME // initialize virtual_time #endif my_init = 1; end_init: netbuf_delete(outbuf); end: return error; } int MPI_Comm_socket(MPI_Comm comm, int *sock) { if (comm != MPI_COMM_WORLD || my_init == 0) return EINVAL; *sock = (int)my_socket.nc; return MPI_SUCCESS; } int MPI_Abort(MPI_Comm comm, int errorcode) { int error; char msg[32]; sprintf(msg, "%d", MINI_MPI_ABORT); if (comm != MPI_COMM_WORLD || my_init == 0) return EINVAL; if ((error = netconn_write(MY_NC, msg, strlen(msg), NETCONN_NOCOPY)) < 0) return error; return MPI_SUCCESS; } int MPI_Finalize(void) { int error; char msg[32]; sprintf(msg, "%d", MINI_MPI_FINAL); if (my_init == 0) return EINVAL; if ((error = netconn_write(MY_NC, msg, strlen(msg), NETCONN_NOCOPY)) < 0) return error; my_init = 0; free(task_array); return MPI_SUCCESS; } /******************************/ int MPI_Send(void *buf, int count, MPI_Datatype type, int dest, int tag, MPI_Comm comm) { int error = 0; struct netconn *sock = (struct netconn*)task_array[dest].sock.nc; int msg_len = (type_size(type) * count) + 1; char msg[msg_len]; if (comm != MPI_COMM_WORLD || my_init == 0 || sock == NULL) return EINVAL; buffer_to_char(msg, msg_len, buf, count, type); #ifdef VIRT_TIME if ((error = vt_send(dest, sock, msg, strlen(msg))) < 0) return error; #else if ((error = netconn_write(sock, msg, strlen(msg), NETCONN_COPY)) < 0) return error; #endif #ifdef MINI_MPI_TRACK { char tracker[msg_len + 33]; sprintf(tracker, "%d:%s", MINI_MPI_SEND, msg); if ((error = netconn_write(MY_NC, tracker, strlen(tracker) + 1, NETCONN_COPY)) < 0) return error; } #endif return MPI_SUCCESS; } int MPI_Recv(void *buf, int count, MPI_Datatype type, int source, int tag, MPI_Comm comm, MPI_Status *status) { char *msg; int msg_len = 0; struct netconn *sock; struct netbuf *inbuf; if (source == -1) sock = bcast_conn; else sock = (struct netconn*)task_array[source].sock.nc; if (comm != MPI_COMM_WORLD || my_init == 0 || sock == NULL) return EINVAL; #ifdef VIRT_TIME msg = vt_recv(sock, MPI_MSG_LEN); char_to_buffer(buf, count, type, msg); #else msg = malloc(MPI_MSG_LEN + 1); inbuf = netconn_recv(sock); if (sock->err != ERR_OK) return sock->err; do { char *data; unsigned short data_len; netbuf_data(inbuf, (void**)&data, &data_len); data[data_len] = '\0'; strncat(msg, data, MPI_MSG_LEN - msg_len); msg_len += data_len; } while (netbuf_next(inbuf) >= 0); char_to_buffer(buf, count, type, msg); #endif #ifdef MINI_MPI_TRACK { int error = 0; char tracker[msg_len + 33]; sprintf(tracker, "%d:%s", MINI_MPI_RECV, msg); if ((error = netconn_write(MY_NC, tracker, strlen(tracker) + 1, NETCONN_COPY)) < 0) return error; } #endif #ifndef VIRT_TIME free(msg); #endif return MPI_SUCCESS; } /******************************/ int MPI_Isend(void *buf, int count, MPI_Datatype type, int dest, int tag, MPI_Comm comm, MPI_Request *request) { int error = 0; struct netconn *sock = (struct netconn*)task_array[dest].sock.nc; int msg_len = (type_size(type) * count) + 1; char msg[msg_len]; if (comm != MPI_COMM_WORLD || my_init == 0 || sock == NULL) return EINVAL; buffer_to_char(msg, msg_len, buf, count, type); #ifdef VIRT_TIME if ((error = vt_send(sock, msg, strlen(msg))) < 0) return error; #else if ((error = netconn_write(sock, msg, strlen(msg), NETCONN_COPY)) < 0) return error; #endif #ifdef MINI_MPI_TRACK { char tracker[msg_len + 33]; sprintf(tracker, "%d:%s", MINI_MPI_SEND, msg); if ((error = netconn_write(MY_NC, tracker, strlen(tracker) + 1, NETCONN_COPY)) < 0) return error; } #endif return MPI_SUCCESS; } int MPI_Irecv(void *buf, int count, MPI_Datatype type, int source, int tag, MPI_Comm comm, MPI_Request *request) { char *msg; int msg_len = 0; struct netconn *sock; struct netbuf *inbuf; if (source == -1) sock = bcast_conn; else sock = (struct netconn*)task_array[source].sock.nc; if (comm != MPI_COMM_WORLD || my_init == 0 || sock == NULL) return EINVAL; #ifdef VIRT_TIME msg = vt_recv(sock, MPI_MSG_LEN); char_to_buffer(buf, count, type, msg); #else msg = malloc(MPI_MSG_LEN + 1); inbuf = netconn_recv(sock); if (sock->err != ERR_OK) return sock->err; do { char *data; unsigned short data_len; netbuf_data(inbuf, (void**)&data, &data_len); data[data_len] = '\0'; strncat(msg, data, MPI_MSG_LEN - msg_len); msg_len += data_len; } while (netbuf_next(inbuf) >= 0); char_to_buffer(buf, count, type, msg); #endif #ifdef MINI_MPI_TRACK { int error = 0; char tracker[msg_len + 33]; sprintf(tracker, "%d:%s", MINI_MPI_RECV, msg); if ((error = netconn_write(MY_NC, tracker, strlen(tracker) + 1, NETCONN_COPY)) < 0) return error; } #endif #ifndef VIRT_TIME free(msg); #endif return MPI_SUCCESS; } int MPI_Wait(MPI_Request *request, MPI_Status *status) { // select? return MPI_SUCCESS; } int MPI_Waitall(int count, MPI_Request *request[], MPI_Status *status[]) { return MPI_SUCCESS; } /******************************/ int MPI_Barrier(MPI_Comm comm) { struct netbuf *resume; int error; unsigned short msg_in_len = 0; char msg_out[32], *msg_in; sprintf(msg_out, "%d", MINI_MPI_BARRIER); if (comm != MPI_COMM_WORLD || my_init == 0 || MY_NC == NULL) return EINVAL; if ((error = netconn_write(MY_NC, msg_out, strlen(msg_out) + 1, NETCONN_NOCOPY)) < 0) return error; await_resume: if ((resume = netconn_recv(MY_NC)) == NULL) return ENOTCONN; netbuf_data(resume, (void**)&msg_in, &msg_in_len); if (atoi(msg_in) != MINI_MPI_RESUME) { // log error goto await_resume; } return MPI_SUCCESS; } int MPI_Bcast(void *buf, int count, MPI_Datatype type, int root, MPI_Comm comm) { int error = 0; int msg_len = (type_size(type) * count) + 1; char msg[msg_len]; if (comm != MPI_COMM_WORLD || my_init == 0 || bcast_conn == NULL) return EINVAL; buffer_to_char(msg, msg_len, buf, count, type); #ifdef VIRT_TIME if ((error = vt_send(bcast_conn, msg, strlen(msg))) < 0) return error; #else if ((error = netconn_write(bcast_conn, msg, strlen(msg), NETCONN_COPY)) < 0) return error; #endif return MPI_SUCCESS; } int MPI_Scatter(void *sendbuffer, int sendcount, MPI_Datatype sendtype, void *recvbuffer, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) { if (comm != MPI_COMM_WORLD || my_init == 0) return EINVAL; #ifdef VIRT_TIME // if ((error = vt_send(MY_NC, msg, strlen(msg))) < 0) // return error; #else // if ((error = netconn_write(MY_NC, msg, strlen(msg), NETCONN_COPY)) < 0) // return error; #endif #ifdef MINI_MPI_TRACK { int error = 0; char tracker[msg_len + 33]; sprintf(tracker, "%d:%s", MINI_MPI_SCATTER, msg); if ((error = netconn_write(MY_NC, tracker, strlen(tracker) + 1, NETCONN_COPY)) < 0) return error; } #endif return MPI_SUCCESS; } int MPI_Gather(void *sendbuffer, int sendcount, MPI_Datatype sendtype, void *recvbuffer, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) { if (comm != MPI_COMM_WORLD || my_init == 0) return EINVAL; #ifdef VIRT_TIME // recv ?? #else #endif #ifdef MINI_MPI_TRACK { int error = 0; char tracker[msg_len + 33]; sprintf(tracker, "%d:%s", MINI_MPI_SEND, msg); if ((error = netconn_write(MY_NC, tracker, strlen(tracker) + 1, NETCONN_COPY)) < 0) return error; } #endif return MPI_SUCCESS; }

© 2007, Los Alamos National Security, LLC.