+++ /dev/null
-/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
- * vim:expandtab:shiftwidth=8:tabstop=8:
- *
- * Copyright (c) 2002 Cray Inc.
- *
- * This file is part of Lustre, http://www.lustre.org.
- *
- * Lustre is free software; you can redistribute it and/or
- * modify it under the terms of version 2 of the GNU General Public
- * License as published by the Free Software Foundation.
- *
- * Lustre is distributed in the hope that 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.
- *
- * You should have received a copy of the GNU General Public License
- * along with Lustre; if not, write to the Free Software
- * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
- */
-
-/* connection.c:
- This file provides a simple stateful connection manager which
- builds tcp connections on demand and leaves them open for
- future use. It also provides the machinery to allow peers
- to connect to it
-*/
-
-#include <stdlib.h>
-#include <pqtimer.h>
-#include <dispatch.h>
-#include <table.h>
-#include <stdio.h>
-#include <stdarg.h>
-#include <string.h>
-#include <unistd.h>
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <netinet/in.h>
-#include <netinet/tcp.h>
-#include <portals/types.h>
-#include <portals/list.h>
-#include <portals/lib-types.h>
-#include <portals/socknal.h>
-#include <linux/kp30.h>
-#include <connection.h>
-#include <pthread.h>
-#include <errno.h>
-#ifndef __CYGWIN__
-#include <syscall.h>
-#endif
-
-/* global variable: acceptor port */
-unsigned short tcpnal_acceptor_port = 988;
-
-
-/* Function: compare_connection
- * Arguments: connection c: a connection in the hash table
- * ptl_process_id_t: an id to verify agains
- * Returns: 1 if the connection is the one requested, 0 otherwise
- *
- * compare_connection() tests for collisions in the hash table
- */
-static int compare_connection(void *arg1, void *arg2)
-{
- connection c = arg1;
- unsigned int * id = arg2;
-#if 0
- return((c->ip==id[0]) && (c->port==id[1]));
-#else
- /* CFS specific hacking */
- return (c->ip == id[0]);
-#endif
-}
-
-
-/* Function: connection_key
- * Arguments: ptl_process_id_t id: an id to hash
- * Returns: a not-particularily-well-distributed hash
- * of the id
- */
-static unsigned int connection_key(unsigned int *id)
-{
-#if 0
- return(id[0]^id[1]);
-#else
- /* CFS specific hacking */
- return (unsigned int) id[0];
-#endif
-}
-
-
-/* Function: remove_connection
- * Arguments: c: the connection to remove
- */
-void remove_connection(void *arg)
-{
- connection c = arg;
- unsigned int id[2];
-
- id[0]=c->ip;
- id[1]=c->port;
- hash_table_remove(c->m->connections,id);
- close(c->fd);
- free(c);
-}
-
-
-/* Function: read_connection:
- * Arguments: c: the connection to read from
- * dest: the buffer to read into
- * len: the number of bytes to read
- * Returns: success as 1, or failure as 0
- *
- * read_connection() reads data from the connection, continuing
- * to read partial results until the request is satisfied or
- * it errors. TODO: this read should be covered by signal protection.
- */
-int read_connection(connection c,
- unsigned char *dest,
- int len)
-{
- int offset = 0,rc;
-
- if (len) {
- do {
-#ifndef __CYGWIN__
- rc = syscall(SYS_read, c->fd, dest+offset, len-offset);
-#else
- rc = recv(c->fd, dest+offset, len-offset, 0);
-#endif
- if (rc <= 0) {
- if (errno == EINTR) {
- rc = 0;
- } else {
- remove_connection(c);
- return (0);
- }
- }
- offset += rc;
- } while (offset < len);
- }
- return (1);
-}
-
-static int connection_input(void *d)
-{
- connection c = d;
- return((*c->m->handler)(c->m->handler_arg,c));
-}
-
-
-/* Function: allocate_connection
- * Arguments: t: tcpnal the allocation is occuring in the context of
- * dest: portal endpoint address for this connection
- * fd: open file descriptor for the socket
- * Returns: an allocated connection structure
- *
- * just encompasses the action common to active and passive
- * connections of allocation and placement in the global table
- */
-static connection allocate_connection(manager m,
- unsigned int ip,
- unsigned short port,
- int fd)
-{
- connection c=malloc(sizeof(struct connection));
- unsigned int id[2];
- c->m=m;
- c->fd=fd;
- c->ip=ip;
- c->port=port;
- id[0]=ip;
- id[1]=port;
- register_io_handler(fd,READ_HANDLER,connection_input,c);
- hash_table_insert(m->connections,c,id);
- return(c);
-}
-
-
-/* Function: new_connection
- * Arguments: t: opaque argument holding the tcpname
- * Returns: 1 in order to reregister for new connection requests
- *
- * called when the bound service socket recieves
- * a new connection request, it always accepts and
- * installs a new connection
- */
-static int new_connection(void *z)
-{
- manager m=z;
- struct sockaddr_in s;
- int len=sizeof(struct sockaddr_in);
- int fd=accept(m->bound,(struct sockaddr *)&s,&len);
- unsigned int nid=*((unsigned int *)&s.sin_addr);
- /* cfs specific hack */
- //unsigned short pid=s.sin_port;
- pthread_mutex_lock(&m->conn_lock);
- allocate_connection(m,htonl(nid),0/*pid*/,fd);
- pthread_mutex_unlock(&m->conn_lock);
- return(1);
-}
-
-/* FIXME assuming little endian, cleanup!! */
-#define __cpu_to_le64(x) ((__u64)(x))
-#define __le64_to_cpu(x) ((__u64)(x))
-#define __cpu_to_le32(x) ((__u32)(x))
-#define __le32_to_cpu(x) ((__u32)(x))
-#define __cpu_to_le16(x) ((__u16)(x))
-#define __le16_to_cpu(x) ((__u16)(x))
-
-extern ptl_nid_t tcpnal_mynid;
-
-int
-tcpnal_hello (int sockfd, ptl_nid_t *nid, int type, __u64 incarnation)
-{
- int rc;
- ptl_hdr_t hdr;
- ptl_magicversion_t *hmv = (ptl_magicversion_t *)&hdr.dest_nid;
-
- LASSERT (sizeof (*hmv) == sizeof (hdr.dest_nid));
-
- memset (&hdr, 0, sizeof (hdr));
- hmv->magic = __cpu_to_le32 (PORTALS_PROTO_MAGIC);
- hmv->version_major = __cpu_to_le32 (PORTALS_PROTO_VERSION_MAJOR);
- hmv->version_minor = __cpu_to_le32 (PORTALS_PROTO_VERSION_MINOR);
-
- hdr.src_nid = __cpu_to_le64 (tcpnal_mynid);
- hdr.type = __cpu_to_le32 (PTL_MSG_HELLO);
-
- hdr.msg.hello.type = __cpu_to_le32 (type);
- hdr.msg.hello.incarnation = 0;
-
- /* Assume sufficient socket buffering for this message */
- rc = syscall(SYS_write, sockfd, &hdr, sizeof(hdr));
- if (rc <= 0) {
- CERROR ("Error %d sending HELLO to "LPX64"\n", rc, *nid);
- return (rc);
- }
-
- rc = syscall(SYS_read, sockfd, hmv, sizeof(*hmv));
- if (rc <= 0) {
- CERROR ("Error %d reading HELLO from "LPX64"\n", rc, *nid);
- return (rc);
- }
-
- if (hmv->magic != __le32_to_cpu (PORTALS_PROTO_MAGIC)) {
- CERROR ("Bad magic %#08x (%#08x expected) from "LPX64"\n",
- __cpu_to_le32 (hmv->magic), PORTALS_PROTO_MAGIC, *nid);
- return (-EPROTO);
- }
-
- if (hmv->version_major != __cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR) ||
- hmv->version_minor != __cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR)) {
- CERROR ("Incompatible protocol version %d.%d (%d.%d expected)"
- " from "LPX64"\n",
- __le16_to_cpu (hmv->version_major),
- __le16_to_cpu (hmv->version_minor),
- PORTALS_PROTO_VERSION_MAJOR,
- PORTALS_PROTO_VERSION_MINOR,
- *nid);
- return (-EPROTO);
- }
-
-#if (PORTALS_PROTO_VERSION_MAJOR != 0)
-# error "This code only understands protocol version 0.x"
-#endif
- /* version 0 sends magic/version as the dest_nid of a 'hello' header,
- * so read the rest of it in now... */
-
- rc = syscall(SYS_read, sockfd, hmv + 1, sizeof(hdr) - sizeof(*hmv));
- if (rc <= 0) {
- CERROR ("Error %d reading rest of HELLO hdr from "LPX64"\n",
- rc, *nid);
- return (rc);
- }
-
- /* ...and check we got what we expected */
- if (hdr.type != __cpu_to_le32 (PTL_MSG_HELLO) ||
- hdr.payload_length != __cpu_to_le32 (0)) {
- CERROR ("Expecting a HELLO hdr with 0 payload,"
- " but got type %d with %d payload from "LPX64"\n",
- __le32_to_cpu (hdr.type),
- __le32_to_cpu (hdr.payload_length), *nid);
- return (-EPROTO);
- }
-
- if (__le64_to_cpu(hdr.src_nid) == PTL_NID_ANY) {
- CERROR("Expecting a HELLO hdr with a NID, but got PTL_NID_ANY\n");
- return (-EPROTO);
- }
-
- if (*nid == PTL_NID_ANY) { /* don't know peer's nid yet */
- *nid = __le64_to_cpu(hdr.src_nid);
- } else if (*nid != __le64_to_cpu (hdr.src_nid)) {
- CERROR ("Connected to nid "LPX64", but expecting "LPX64"\n",
- __le64_to_cpu (hdr.src_nid), *nid);
- return (-EPROTO);
- }
-
- return (0);
-}
-
-/* Function: force_tcp_connection
- * Arguments: t: tcpnal
- * dest: portals endpoint for the connection
- * Returns: an allocated connection structure, either
- * a pre-existing one, or a new connection
- */
-connection force_tcp_connection(manager m,
- unsigned int ip,
- unsigned short port,
- procbridge pb)
-{
- connection conn;
- struct sockaddr_in addr;
- unsigned int id[2];
-
- port = tcpnal_acceptor_port;
-
- id[0] = ip;
- id[1] = port;
-
- pthread_mutex_lock(&m->conn_lock);
-
- conn = hash_table_find(m->connections, id);
- if (!conn) {
- int fd;
- int option;
- ptl_nid_t peernid = PTL_NID_ANY;
-
- bzero((char *) &addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_addr.s_addr = htonl(ip);
- addr.sin_port = htons(port);
-
- if ((fd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
- perror("tcpnal socket failed");
- exit(-1);
- }
- if (connect(fd, (struct sockaddr *)&addr,
- sizeof(struct sockaddr_in))) {
- perror("tcpnal connect");
- return(0);
- }
-
-#if 1
- option = 1;
- setsockopt(fd, SOL_TCP, TCP_NODELAY, &option, sizeof(option));
- option = 1<<20;
- setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &option, sizeof(option));
- option = 1<<20;
- setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &option, sizeof(option));
-#endif
-
- /* say hello */
- if (tcpnal_hello(fd, &peernid, SOCKNAL_CONN_ANY, 0))
- exit(-1);
-
- conn = allocate_connection(m, ip, port, fd);
-
- /* let nal thread know this event right away */
- if (conn)
- procbridge_wakeup_nal(pb);
- }
-
- pthread_mutex_unlock(&m->conn_lock);
- return (conn);
-}
-
-
-/* Function: bind_socket
- * Arguments: t: the nal state for this interface
- * port: the port to attempt to bind to
- * Returns: 1 on success, or 0 on error
- *
- * bind_socket() attempts to allocate and bind a socket to the requested
- * port, or dynamically assign one from the kernel should the port be
- * zero. Sets the bound and bound_handler elements of m.
- *
- * TODO: The port should be an explicitly sized type.
- */
-static int bind_socket(manager m,unsigned short port)
-{
- struct sockaddr_in addr;
- int alen=sizeof(struct sockaddr_in);
-
- if ((m->bound = socket(AF_INET, SOCK_STREAM, 0)) < 0)
- return(0);
-
- bzero((char *) &addr, sizeof(addr));
- addr.sin_family = AF_INET;
- addr.sin_addr.s_addr = 0;
- addr.sin_port = htons(port);
-
- if (bind(m->bound,(struct sockaddr *)&addr,alen)<0){
- perror ("tcpnal bind");
- return(0);
- }
-
- getsockname(m->bound,(struct sockaddr *)&addr, &alen);
-
- m->bound_handler=register_io_handler(m->bound,READ_HANDLER,
- new_connection,m);
- listen(m->bound,5);
- m->port=addr.sin_port;
- return(1);
-}
-
-
-/* Function: shutdown_connections
- * Arguments: m: the manager structure
- *
- * close all connections and reclaim resources
- */
-void shutdown_connections(manager m)
-{
- close(m->bound);
- remove_io_handler(m->bound_handler);
- hash_destroy_table(m->connections,remove_connection);
- free(m);
-}
-
-
-/* Function: init_connections
- * Arguments: t: the nal state for this interface
- * port: the port to attempt to bind to
- * Returns: a newly allocated manager structure, or
- * zero if the fixed port could not be bound
- */
-manager init_connections(unsigned short pid,
- int (*input)(void *, void *),
- void *a)
-{
- manager m = (manager)malloc(sizeof(struct manager));
- m->connections = hash_create_table(compare_connection,connection_key);
- m->handler = input;
- m->handler_arg = a;
- pthread_mutex_init(&m->conn_lock, 0);
-
- if (bind_socket(m,pid))
- return(m);
-
- free(m);
- return(0);
-}
git://git.whamcloud.com / fs / lustre-release.git / blobdiff