/*

  This file is provided under a dual BSD/GPLv2 license.  When using or

  redistributing this file, you may do so under either license.

  GPL LICENSE SUMMARY

  Copyright(c) 2016 Intel Corporation.

  This program 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.

  This program 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.

  Contact Information:

  Intel Corporation, www.intel.com

  BSD LICENSE

  Copyright(c) 2016 Intel Corporation.

  Redistribution and use in source and binary forms, with or without

  modification, are permitted provided that the following conditions

  are met:

    * Redistributions of source code must retain the above copyright

      notice, this list of conditions and the following disclaimer.

    * 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.

    * Neither the name of Intel Corporation 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 COPYRIGHT HOLDERS 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 COPYRIGHT

  OWNER 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.

*/

/* Copyright (c) 2003-2016 Intel Corporation. All rights reserved. */

#include <sys/types.h>		/* shm_open and signal handling */

#include <sys/mman.h>

#include <sys/stat.h>

#include <fcntl.h>

#include <signal.h>

#include "psm_user.h"

#include "psm_mq_internal.h"

#include "psm_am_internal.h"

#include "cmarw.h"

#include "psmi_wrappers.h"

#ifdef PSM_CUDA

#include "am_cuda_memhandle_cache.h"

#endif

int psmi_shm_mq_rv_thresh = PSMI_MQ_RV_THRESH_NO_KASSIST;

static const amsh_qinfo_t amsh_qcounts = {

	.qreqFifoShort = 1024,

	.qreqFifoLong = 256,

	.qrepFifoShort = 1024,

	.qrepFifoLong = 256

};

static const amsh_qinfo_t amsh_qelemsz = {

	.qreqFifoShort = sizeof(am_pkt_short_t),

	.qreqFifoLong = AMLONG_SZ,

	.qrepFifoShort = sizeof(am_pkt_short_t),

	.qrepFifoLong = AMLONG_SZ

};

ustatic struct {

	void *addr;

	size_t len;

	struct sigaction SIGSEGV_old_act;

	struct sigaction SIGBUS_old_act;

} action_stash;

static psm2_error_t amsh_poll(ptl_t *ptl, int replyonly);

static void process_packet(ptl_t *ptl, am_pkt_short_t *pkt, int isreq);

static void amsh_conn_handler(void *toki, psm2_amarg_t *args, int narg,

			      void *buf, size_t len);

/* Kassist helper functions */

#if _HFI_DEBUGGING

static const char *psmi_kassist_getmode(int mode);

#endif

static int psmi_get_kassist_mode();

int psmi_epaddr_pid(psm2_epaddr_t epaddr);

static inline void

am_ctl_qhdr_init(volatile am_ctl_qhdr_t *q, int elem_cnt, int elem_sz)

{

	pthread_spin_init(&q->lock, PTHREAD_PROCESS_SHARED);

	q->head = 0;

	q->tail = 0;

	q->elem_cnt = elem_cnt;

	q->elem_sz = elem_sz;

}

static void

am_ctl_bulkpkt_init(am_pkt_bulk_t *base_ptr, size_t elemsz, int nelems)

{

	int i;

	am_pkt_bulk_t *bulkpkt;

	uintptr_t bulkptr = (uintptr_t) base_ptr;

	for (i = 0; i < nelems; i++, bulkptr += elemsz) {

		bulkpkt = (am_pkt_bulk_t *) bulkptr;

		bulkpkt->idx = i;

	}

}

#define _PA(type) PSMI_ALIGNUP(amsh_qcounts.q ## type * amsh_qelemsz.q ## type, \

			       PSMI_PAGESIZE)

static inline uintptr_t am_ctl_sizeof_block()

{

	return PSMI_ALIGNUP(

			PSMI_ALIGNUP(AMSH_BLOCK_HEADER_SIZE, PSMI_PAGESIZE) +

			/* reqctrl block */

			PSMI_ALIGNUP(sizeof(am_ctl_blockhdr_t), PSMI_PAGESIZE) +

			_PA(reqFifoShort) + _PA(reqFifoLong) +

			/*reqctrl block */

			PSMI_ALIGNUP(sizeof(am_ctl_blockhdr_t), PSMI_PAGESIZE) +

			/* align to page size */

			_PA(repFifoShort) + _PA(repFifoLong), PSMI_PAGESIZE);

}

#undef _PA

static uint32_t create_extra_ep_data()

{

	uint32_t ret = getpid();

#ifdef PSM_CUDA

	/* PID is at maximum 22 bits */

	ret |= my_gpu_device << 22;

#endif

	return ret;

}

static void read_extra_ep_data(uint32_t data, uint32_t *pid, uint32_t *gpu)

{

	uint32_t pid_mask = (1 << 22) - 1;

	*pid = data & pid_mask;

	*gpu = (data & ~pid_mask) >> 22;

}

static void am_update_directory(struct am_ctl_nodeinfo *);

static

void amsh_atexit()

{

	static ips_atomic_t atexit_once = { 0 };

	psm2_ep_t ep;

	struct ptl_am *ptl;

	/* bail out if previous value is non-zero */

	if (ips_atomic_cmpxchg(&atexit_once, 0, 1) != 0)

		return;

	ep = psmi_opened_endpoint;

	while (ep) {

		ptl = (struct ptl_am *)(ep->ptl_amsh.ptl);

		if (ptl->self_nodeinfo &&

		    ptl->amsh_keyname != NULL) {

			_HFI_VDBG("unlinking shm file %s\n",

				  ptl->amsh_keyname);

			shm_unlink(ptl->amsh_keyname);

		}

		ep = ep->user_ep_next;

	}

	return;

}

ustatic

void amsh_mmap_fault(int signo, siginfo_t *siginfo, void *context)

{

	if ((unsigned long int) siginfo->si_addr >= (unsigned long int) action_stash.addr &&

	    (unsigned long int) siginfo->si_addr <  (unsigned long int) action_stash.addr + (unsigned long int) action_stash.len) {

		static char shm_errmsg[256];

		snprintf(shm_errmsg, sizeof(shm_errmsg),

			 "%s: Unable to allocate shared memory for intra-node messaging.\n"

			 "%s: Delete stale shared memory files in /dev/shm.\n",

			 psmi_gethostname(), psmi_gethostname());

		amsh_atexit();

		if (psmi_write(2, shm_errmsg, strlen(shm_errmsg) + 1) == -1)

			psmi_exit(2);

		else

			psmi_exit(1); /* XXX revisit this... there's probably a better way to exit */

	} else {

		if (signo == SIGSEGV) {

			if (action_stash.SIGSEGV_old_act.sa_sigaction == (void*) SIG_DFL) {

				psmi_sigaction(SIGSEGV, &action_stash.SIGSEGV_old_act, NULL);

				raise(SIGSEGV);

				struct sigaction act;

				act.sa_sigaction = amsh_mmap_fault;

				act.sa_flags = SA_SIGINFO;

				psmi_sigaction(SIGSEGV, &act, NULL);

			} else if (action_stash.SIGSEGV_old_act.sa_sigaction == (void*) SIG_IGN) {

				return;

			} else {

				action_stash.SIGSEGV_old_act.sa_sigaction(signo, siginfo, context);

			}

		} else if (signo == SIGBUS) {

			if (action_stash.SIGBUS_old_act.sa_sigaction == (void*) SIG_DFL) {

				psmi_sigaction(SIGBUS, &action_stash.SIGBUS_old_act, NULL);

				raise(SIGBUS);

				struct sigaction act;

				act.sa_sigaction = amsh_mmap_fault;

				act.sa_flags = SA_SIGINFO;

				psmi_sigaction(SIGBUS, &act, NULL);

			} else if (action_stash.SIGBUS_old_act.sa_sigaction == (void*) SIG_IGN) {

				return;

			} else {

				action_stash.SIGBUS_old_act.sa_sigaction(signo, siginfo, context);

			}

		} else {

			psmi_exit(signo);

		}

	}

}

/**

 * Create endpoint shared-memory object, containing ep's info

 * and message queues.

 */

psm2_error_t psmi_shm_create(ptl_t *ptl_gen)

{

	struct ptl_am *ptl = (struct ptl_am *)ptl_gen;

	psm2_ep_t ep = ptl->ep;

	char shmbuf[256];

	void *mapptr;

	size_t segsz;

	psm2_error_t err = PSM2_OK;

	int shmfd = -1;

	char *amsh_keyname = NULL;

	int iterator;

	/* Get which kassist mode to use. */

	ptl->psmi_kassist_mode = psmi_get_kassist_mode();

	if (_HFI_PRDBG_ON) {

		_HFI_PRDBG_ALWAYS

			("kassist_mode %d %s use_kassist %d\n",

			ptl->psmi_kassist_mode,

			psmi_kassist_getmode(ptl->psmi_kassist_mode),

			(ptl->psmi_kassist_mode != PSMI_KASSIST_OFF));

	}

	segsz = am_ctl_sizeof_block();

	for (iterator = 0; iterator <= INT_MAX; iterator++) {

		snprintf(shmbuf,

			 sizeof(shmbuf),

			 "/psm2_shm.%ld%016lx%d",

			 (long int) getuid(),

			 ep->epid,

			 iterator);

		amsh_keyname = psmi_strdup(NULL, shmbuf);

		if (amsh_keyname == NULL) {

			err = PSM2_NO_MEMORY;

			goto fail;

		}

		shmfd =

		    shm_open(amsh_keyname, O_RDWR | O_CREAT, S_IRUSR | S_IWUSR);

		if (shmfd < 0) {

			psmi_free(amsh_keyname);

			amsh_keyname = NULL;

			if (errno == EACCES && iterator < INT_MAX)

				continue;

			else {

				err = psmi_handle_error(NULL,

							PSM2_SHMEM_SEGMENT_ERR,

							"Error creating shared "

							"memory object in "

							"shm_open: %s",

							strerror(errno));

				goto fail;

			}

		} else {

			struct stat st;

			if (fstat(shmfd, &st) == -1) {

				err = psmi_handle_error(NULL,

							PSM2_SHMEM_SEGMENT_ERR,

							"Error validating "

							"shared memory object "

							"with fstat: %s",

							strerror(errno));

				goto fail;

			}

			if (getuid() == st.st_uid) {

				err = PSM2_OK;

				break;

			} else {

				err = PSM2_SHMEM_SEGMENT_ERR;

				close(shmfd);

			}

		}

	}

	if (err) {

		if (amsh_keyname) psmi_free(amsh_keyname);

		err = psmi_handle_error(NULL,

					PSM2_SHMEM_SEGMENT_ERR,

					"Error creating shared memory object "

					"in shm_open: namespace exhausted.");

		goto fail;

	}

	/* Now register the atexit handler for cleanup, whether master or slave */

	atexit(amsh_atexit);

	_HFI_PRDBG("Opened shmfile %s\n", amsh_keyname);

	if (ftruncate(shmfd, segsz) != 0) {

		err = psmi_handle_error(NULL, PSM2_SHMEM_SEGMENT_ERR,

					"Error setting size of shared memory object to %u bytes in "

					"ftruncate: %s\n",

					(uint32_t) segsz,

					strerror(errno));

		goto fail;

	}

	mapptr = mmap(NULL, segsz,

		      PROT_READ | PROT_WRITE, MAP_SHARED, shmfd, 0);

	if (mapptr == MAP_FAILED) {

		err = psmi_handle_error(NULL, PSM2_SHMEM_SEGMENT_ERR,

					"Error mmapping shared memory: %s",

					strerror(errno));

		psmi_free(amsh_keyname);

		goto fail;

	}

	memset((void *) mapptr, 0, segsz); /* touch all of my pages */

	/* Our own ep's info for ptl_am resides at the start of the

	   shm object.  Other processes need some of this info to

	   understand the rest of the queue structure and other details. */

	ptl->self_nodeinfo = (struct am_ctl_nodeinfo *) mapptr;

	ptl->amsh_keyname = amsh_keyname;

	ptl->self_nodeinfo->amsh_shmbase = (uintptr_t) mapptr;

fail:

	if (shmfd >= 0) close(shmfd);

	return err;

}

psm2_error_t psmi_epdir_extend(ptl_t *ptl_gen)

{

	struct ptl_am *ptl = (struct ptl_am *)ptl_gen;

	struct am_ctl_nodeinfo *new = NULL;

	new = (struct am_ctl_nodeinfo *)

		psmi_memalign(ptl->ep, PER_PEER_ENDPOINT, 64,

			      (ptl->am_ep_size + AMSH_DIRBLOCK_SIZE) *

			      sizeof(struct am_ctl_nodeinfo));

	if (new == NULL)

		return PSM2_NO_MEMORY;

	memcpy(new, ptl->am_ep,

	       ptl->am_ep_size * sizeof(struct am_ctl_nodeinfo));

	memset(new + ptl->am_ep_size, 0,

	       AMSH_DIRBLOCK_SIZE * sizeof(struct am_ctl_nodeinfo));

	psmi_free(ptl->am_ep);

	ptl->am_ep = new;

	ptl->am_ep_size += AMSH_DIRBLOCK_SIZE;

	return PSM2_OK;

}

/**

 * Unmap shm regions upon proper disconnect with other processes

 */

psm2_error_t psmi_do_unmap(uintptr_t shmbase)

{

	psm2_error_t err = PSM2_OK;

	if (munmap((void *)shmbase, am_ctl_sizeof_block())) {

		err =

		    psmi_handle_error(NULL, PSM2_SHMEM_SEGMENT_ERR,

				      "Error with munmap of shared segment: %s",

				      strerror(errno));

	}

	return err;

}

/**

 * Map a remote process' shared memory object.

 *

 * If the remote process has a shared memory object available, add it to our own

 * directory and return the shmidx.  If the shared memory object does not exist,

 * return -1, and the connect poll function will try to map again later.

 *

 * If force_remap is true, then clear the entry that matches the epid.

 */

psm2_error_t psmi_shm_map_remote(ptl_t *ptl_gen, psm2_epid_t epid, uint16_t *shmidx_o, int force_remap)

{

	struct ptl_am *ptl = (struct ptl_am *)ptl_gen;

	int i;

	int use_kassist;

	uint16_t shmidx;

	char shmbuf[256];

	void *dest_mapptr;

	size_t segsz;

	psm2_error_t err = PSM2_OK;

	int dest_shmfd;

	struct am_ctl_nodeinfo *dest_nodeinfo;

	int iterator;

	shmidx = *shmidx_o = -1;

	for (i = 0; i <= ptl->max_ep_idx; i++) {

		if (ptl->am_ep[i].epid == epid) {

			if (force_remap) {

				ptl->am_ep[i].epaddr = NULL;

				ptl->am_ep[i].epid = 0;

				break;

			}

			*shmidx_o = shmidx = i;

			return err;

		}

	}

	use_kassist = (ptl->psmi_kassist_mode != PSMI_KASSIST_OFF);

	segsz = am_ctl_sizeof_block();

	for (iterator = 0; iterator <= INT_MAX; iterator++) {

		snprintf(shmbuf,

			 sizeof(shmbuf),

			 "/psm2_shm.%ld%016lx%d",

			 (long int) getuid(),

			 epid,

			 iterator);

		dest_shmfd = shm_open(shmbuf, O_RDWR, S_IRWXU);

		if (dest_shmfd < 0) {

			if (errno == EACCES && iterator < INT_MAX)

				continue;

			else {

				err = psmi_handle_error(NULL,

							PSM2_SHMEM_SEGMENT_ERR,

							"Error opening remote "

							"shared memory object "

							"in shm_open: %s",

							strerror(errno));

				goto fail;

			}

		} else {

			struct stat st;

			if (fstat(dest_shmfd, &st) == -1) {

				err = psmi_handle_error(NULL,

							PSM2_SHMEM_SEGMENT_ERR,

							"Error validating "

							"shared memory object "

							"with fstat: %s",

							strerror(errno));

				close(dest_shmfd);

				goto fail;

			}

			if (getuid() == st.st_uid) {

				err = PSM2_OK;

				break;

			} else {

				err = PSM2_SHMEM_SEGMENT_ERR;

				close(dest_shmfd);

			}

		}

	}

	if (err) {

		err = psmi_handle_error(NULL,

					PSM2_SHMEM_SEGMENT_ERR,

					"Error opening remote shared "

					"memory object in shm_open: "

					"namespace exhausted.");

		goto fail;

	}

	dest_mapptr = mmap(NULL, segsz,

		      PROT_READ | PROT_WRITE, MAP_SHARED, dest_shmfd, 0);

	if (dest_mapptr == MAP_FAILED) {

		err = psmi_handle_error(NULL, PSM2_SHMEM_SEGMENT_ERR,

					"Error mmapping remote shared memory: %s",

					strerror(errno));

		close(dest_shmfd);

		goto fail;

	}

	close(dest_shmfd);

	dest_nodeinfo = (struct am_ctl_nodeinfo *)dest_mapptr;

	/* We core dump right after here if we don't check the mmap */

	action_stash.addr = dest_mapptr;

	action_stash.len = segsz;

	struct sigaction act = { .sa_sigaction = amsh_mmap_fault, .sa_flags = SA_SIGINFO };

	sigaction(SIGSEGV, &act, &action_stash.SIGSEGV_old_act);

	sigaction(SIGBUS, &act, &action_stash.SIGBUS_old_act);

	{

		volatile uint16_t *is_init = &dest_nodeinfo->is_init;

		while (*is_init == 0)

			usleep(1);

		ips_sync_reads();

		_HFI_PRDBG("Got a published remote dirpage page at "

			   "%p, size=%dn", dest_mapptr, (int)segsz);

	}

	shmidx = -1;

	if ((ptl->max_ep_idx + 1) == ptl->am_ep_size) {

		err = psmi_epdir_extend(ptl_gen);

		if (err)

			goto fail;

		for (i = 0; i <= ptl->max_ep_idx; i++) {

			if (ptl->am_ep[i].epid != 0)

				am_update_directory(&ptl->am_ep[i]);

		}

	}

	for (i = 0; i < ptl->am_ep_size; i++) {

		psmi_assert(ptl->am_ep[i].epid != epid);

		if (ptl->am_ep[i].epid == 0) {

			ptl->am_ep[i].epid = epid;

			ptl->am_ep[i].psm_verno = dest_nodeinfo->psm_verno;

			ptl->am_ep[i].pid = dest_nodeinfo->pid;

			if (use_kassist) {

				/* If we are able to use CMA assume everyone

				 * else on the node can also use it.

				 * Advertise that CMA is active via the

				 * feature flag.

				 */

				if (cma_available()) {

					ptl->am_ep[i].amsh_features |=

					    AMSH_HAVE_CMA;

					psmi_shm_mq_rv_thresh =

					    PSMI_MQ_RV_THRESH_CMA;

				} else {

					ptl->psmi_kassist_mode =

					    PSMI_KASSIST_OFF;

					use_kassist = 0;

					psmi_shm_mq_rv_thresh =

					    PSMI_MQ_RV_THRESH_NO_KASSIST;

				}

			} else

				psmi_shm_mq_rv_thresh =

				    PSMI_MQ_RV_THRESH_NO_KASSIST;

			_HFI_PRDBG("KASSIST MODE: %s\n",

				   psmi_kassist_getmode(ptl->psmi_kassist_mode));

			shmidx = *shmidx_o = i;

			_HFI_PRDBG("Mapped epid %lx into shmidx %d\n", epid, shmidx);

			ptl->am_ep[i].amsh_shmbase = (uintptr_t) dest_mapptr;

			ptl->am_ep[i].amsh_qsizes = dest_nodeinfo->amsh_qsizes;

			if (i > ptl->max_ep_idx)

				ptl->max_ep_idx = i;

			break;

		}

	}

	/* install the old sighandler back */

	sigaction(SIGSEGV, &action_stash.SIGSEGV_old_act, NULL);

	sigaction(SIGBUS, &action_stash.SIGBUS_old_act, NULL);

	if (shmidx == (uint16_t)-1)

		err = psmi_handle_error(NULL, PSM2_SHMEM_SEGMENT_ERR,

					"Could not connect to local endpoint");

fail:

	return err;

}

/**

 * Initialize pointer structure and locks for endpoint shared-memory AM.

 */

#define AMSH_QSIZE(type)                                                \

	PSMI_ALIGNUP(amsh_qelemsz.q ## type * amsh_qcounts.q ## type,   \

		     PSMI_PAGESIZE)

static psm2_error_t amsh_init_segment(ptl_t *ptl_gen)

{

	struct ptl_am *ptl = (struct ptl_am *)ptl_gen;

	psm2_error_t err = PSM2_OK;

	/* Preconditions */

	psmi_assert_always(ptl != NULL);

	psmi_assert_always(ptl->ep != NULL);

	psmi_assert_always(ptl->epaddr != NULL);

	psmi_assert_always(ptl->ep->epid != 0);

	if ((err = psmi_shm_create(ptl_gen)))

		goto fail;

	ptl->self_nodeinfo->amsh_qsizes.qreqFifoShort = AMSH_QSIZE(reqFifoShort);

	ptl->self_nodeinfo->amsh_qsizes.qreqFifoLong = AMSH_QSIZE(reqFifoLong);

	ptl->self_nodeinfo->amsh_qsizes.qrepFifoShort = AMSH_QSIZE(repFifoShort);

	ptl->self_nodeinfo->amsh_qsizes.qrepFifoLong = AMSH_QSIZE(repFifoLong);

	/* We core dump right after here if we don't check the mmap */

	struct sigaction act = {

		.sa_sigaction = amsh_mmap_fault,

		.sa_flags = SA_SIGINFO

	};

	sigaction(SIGSEGV, &act, &action_stash.SIGSEGV_old_act);

	sigaction(SIGBUS, &act, &action_stash.SIGBUS_old_act);

	/*

	 * Now that we know our epid, update it in the shmidx array

	 */

	ptl->reqH.base = ptl->reqH.head = ptl->reqH.end = NULL;

	ptl->repH.base = ptl->repH.head = ptl->repH.end = NULL;

	am_update_directory(ptl->self_nodeinfo);

	ptl->reqH.head = ptl->reqH.base = (am_pkt_short_t *)

		(((uintptr_t)ptl->self_nodeinfo->qdir.qreqFifoShort));

	ptl->reqH.end = (am_pkt_short_t *)

		(((uintptr_t)ptl->self_nodeinfo->qdir.qreqFifoShort) +

		 amsh_qcounts.qreqFifoShort * amsh_qelemsz.qreqFifoShort);

	ptl->repH.head = ptl->repH.base = (am_pkt_short_t *)

		(((uintptr_t)ptl->self_nodeinfo->qdir.qrepFifoShort));

	ptl->repH.end = (am_pkt_short_t *)

		(((uintptr_t)ptl->self_nodeinfo->qdir.qrepFifoShort) +

		 amsh_qcounts.qrepFifoShort * amsh_qelemsz.qrepFifoShort);

	am_ctl_qhdr_init(&ptl->self_nodeinfo->qdir.qreqH->shortq,

			 amsh_qcounts.qreqFifoShort,

			 amsh_qelemsz.qreqFifoShort);

	am_ctl_qhdr_init(&ptl->self_nodeinfo->qdir.qreqH->longbulkq,

			 amsh_qcounts.qreqFifoLong, amsh_qelemsz.qreqFifoLong);

	am_ctl_qhdr_init(&ptl->self_nodeinfo->qdir.qrepH->shortq,

			 amsh_qcounts.qrepFifoShort,

			 amsh_qelemsz.qrepFifoShort);

	am_ctl_qhdr_init(&ptl->self_nodeinfo->qdir.qrepH->longbulkq,

			 amsh_qcounts.qrepFifoLong, amsh_qelemsz.qrepFifoLong);

	/* Set bulkidx in every bulk packet */

	am_ctl_bulkpkt_init(ptl->self_nodeinfo->qdir.qreqFifoLong,

			    amsh_qelemsz.qreqFifoLong,

			    amsh_qcounts.qreqFifoLong);

	am_ctl_bulkpkt_init(ptl->self_nodeinfo->qdir.qrepFifoLong,

			    amsh_qelemsz.qrepFifoLong,

			    amsh_qcounts.qrepFifoLong);

	/* install the old sighandler back */

	sigaction(SIGSEGV, &action_stash.SIGSEGV_old_act, NULL);

	sigaction(SIGBUS, &action_stash.SIGBUS_old_act, NULL);

fail:

	return err;

}

psm2_error_t psmi_shm_detach(ptl_t *ptl_gen)

{

	struct ptl_am *ptl = (struct ptl_am *)ptl_gen;

	psm2_error_t err = PSM2_OK;

	uintptr_t shmbase;

	if (ptl->self_nodeinfo == NULL)

		return err;

	_HFI_VDBG("unlinking shm file %s\n", ptl->amsh_keyname + 1);

	shmbase = ptl->self_nodeinfo->amsh_shmbase;

	shm_unlink(ptl->amsh_keyname);

	psmi_free(ptl->amsh_keyname);

	if (munmap((void *)shmbase, am_ctl_sizeof_block())) {

		err =

		    psmi_handle_error(NULL, PSM2_SHMEM_SEGMENT_ERR,

				      "Error with munmap of shared segment: %s",

				      strerror(errno));

		goto fail;

	}

	ptl->self_nodeinfo = NULL;

	return PSM2_OK;

fail:

	return err;

}

/**

 * Update locally shared-pointer directory.  The directory must be

 * updated when a new epaddr is connected to or on every epaddr already

 * connected to whenever the shared memory segment is relocated via mremap.

 *

 * @param epaddr Endpoint address for which to update local directory.

 */

static

void am_update_directory(struct am_ctl_nodeinfo *nodeinfo)

{

	uintptr_t base_this;

	base_this = nodeinfo->amsh_shmbase +

		AMSH_BLOCK_HEADER_SIZE;

	/* Request queues */

	nodeinfo->qdir.qreqH = (am_ctl_blockhdr_t *) base_this;

	nodeinfo->qdir.qreqFifoShort = (am_pkt_short_t *)

	    ((uintptr_t) nodeinfo->qdir.qreqH +

	     PSMI_ALIGNUP(sizeof(am_ctl_blockhdr_t), PSMI_PAGESIZE));

	nodeinfo->qdir.qreqFifoLong = (am_pkt_bulk_t *)

	    ((uintptr_t) nodeinfo->qdir.qreqFifoShort +

	     nodeinfo->amsh_qsizes.qreqFifoShort);

	/* Reply queues */

	nodeinfo->qdir.qrepH = (am_ctl_blockhdr_t *)

	    ((uintptr_t) nodeinfo->qdir.qreqFifoLong +

	     nodeinfo->amsh_qsizes.qreqFifoLong);

	nodeinfo->qdir.qrepFifoShort = (am_pkt_short_t *)

	    ((uintptr_t) nodeinfo->qdir.qrepH +

	     PSMI_ALIGNUP(sizeof(am_ctl_blockhdr_t), PSMI_PAGESIZE));

	nodeinfo->qdir.qrepFifoLong = (am_pkt_bulk_t *)

	    ((uintptr_t) nodeinfo->qdir.qrepFifoShort +

	     nodeinfo->amsh_qsizes.qrepFifoShort);

	_HFI_VDBG("epaddr=%p Request Hdr=%p,Pkt=%p,Long=%p\n",

		  nodeinfo->epaddr,

		  nodeinfo->qdir.qreqH,

		  nodeinfo->qdir.qreqFifoShort,

		  nodeinfo->qdir.qreqFifoLong);

	_HFI_VDBG("epaddr=%p Reply   Hdr=%p,Pkt=%p,Long=%p\n",

		  nodeinfo->epaddr,

		  nodeinfo->qdir.qrepH,

		  nodeinfo->qdir.qrepFifoShort,

		  nodeinfo->qdir.qrepFifoLong);

	/* Sanity check */

	uintptr_t base_next =

	    (uintptr_t) nodeinfo->qdir.qrepFifoLong +

	    nodeinfo->amsh_qsizes.qrepFifoLong;

	psmi_assert_always(base_next - base_this <= am_ctl_sizeof_block());

}

/* ep_epid_share_memory wrapper */

static

int amsh_epid_reachable(ptl_t *ptl_gen, psm2_epid_t epid)

{

	struct ptl_am *ptl = (struct ptl_am *)ptl_gen;

	int result;

	psm2_error_t err;

	err = psm2_ep_epid_share_memory(ptl->ep, epid, &result);

	psmi_assert_always(err == PSM2_OK);

	return result;

}

static

psm2_error_t

amsh_epaddr_add(ptl_t *ptl_gen, psm2_epid_t epid, uint16_t shmidx, psm2_epaddr_t *epaddr_o)

{

	struct ptl_am *ptl = (struct ptl_am *)ptl_gen;

	psm2_epaddr_t epaddr;

	am_epaddr_t *amaddr;

	psm2_error_t err = PSM2_OK;

	psmi_assert(psmi_epid_lookup(ptl->ep, epid) == NULL);

	/* The self PTL handles loopback communication. */

	psmi_assert(epid != ptl->epid);

	/* note the size of the memory is am_epaddr_t */

	epaddr = (psm2_epaddr_t) psmi_calloc(ptl->ep,

					    PER_PEER_ENDPOINT, 1,

					    sizeof(am_epaddr_t));

	if (epaddr == NULL) {

		return PSM2_NO_MEMORY;

	}

	psmi_assert_always(ptl->am_ep[shmidx].epaddr == NULL);

	if ((err = psmi_epid_set_hostname(psm2_epid_nid(epid),

					  psmi_gethostname(), 0)))

		goto fail;

	epaddr->ptlctl = ptl->ctl;

	epaddr->epid = epid;

	/* convert to am_epaddr_t */

	amaddr = (am_epaddr_t *) epaddr;

	/* tell the other endpoint their location in our directory */

	amaddr->shmidx = shmidx;

	/* we haven't connected yet, so we can't give them the same hint */

	amaddr->return_shmidx = -1;

	amaddr->cstate_outgoing = AMSH_CSTATE_OUTGOING_NONE;

	amaddr->cstate_incoming = AMSH_CSTATE_INCOMING_NONE;

	/* other setup */

	ptl->am_ep[shmidx].epaddr = epaddr;

	am_update_directory(&ptl->am_ep[shmidx]);

	/* Finally, add to table */

	if ((err = psmi_epid_add(ptl->ep, epid, epaddr)))

		goto fail;

	_HFI_VDBG("epaddr=%s added to ptl=%p\n",

		  psmi_epaddr_get_name(epid), ptl);

	*epaddr_o = epaddr;

	return PSM2_OK;

fail:

	if (epaddr != ptl->epaddr)

		psmi_free(epaddr);

	return err;

}

static

void

amsh_epaddr_update(ptl_t *ptl_gen, psm2_epaddr_t epaddr)

{

	struct ptl_am *ptl = (struct ptl_am *)ptl_gen;

	am_epaddr_t *amaddr;

	uint16_t shmidx;

	struct am_ctl_nodeinfo *nodeinfo;

	amaddr = (am_epaddr_t *) epaddr;

	shmidx = amaddr->shmidx;

	nodeinfo = (struct am_ctl_nodeinfo *) ptl->am_ep[shmidx].amsh_shmbase;

	/* restart the connection process */

	amaddr->return_shmidx = -1;

	amaddr->cstate_outgoing = AMSH_CSTATE_OUTGOING_NONE;

	/* wait for the other process to init again */

	{

		volatile uint16_t *is_init = &nodeinfo->is_init;

		while (*is_init == 0)

			usleep(1);

		ips_sync_reads();

	}

	/* get the updated values from the new nodeinfo page */

	ptl->am_ep[shmidx].psm_verno = nodeinfo->psm_verno;

	ptl->am_ep[shmidx].pid = nodeinfo->pid;

	ptl->am_ep[shmidx].amsh_qsizes = nodeinfo->amsh_qsizes;

	am_update_directory(&ptl->am_ep[shmidx]);

	return;

}

struct ptl_connection_req {

	int isdone;

	int op;			/* connect or disconnect */

	int numep;

	int numep_left;

	int phase;

	int *epid_mask;

	const psm2_epid_t *epids;	/* input epid list */

	psm2_epaddr_t *epaddr;

	psm2_error_t *errors;	/* inout errors */

	/* Used for connect/disconnect */

	psm2_amarg_t args[4];

};

static

void amsh_free_epaddr(psm2_epaddr_t epaddr)

{

	psmi_epid_remove(epaddr->ptlctl->ep, epaddr->epid);

	psmi_free(epaddr);

	return;