diff options
Diffstat (limited to 'arch/arm/mvp/commkm/comm.c')
| -rw-r--r-- | arch/arm/mvp/commkm/comm.c | 1457 |
1 files changed, 0 insertions, 1457 deletions
diff --git a/arch/arm/mvp/commkm/comm.c b/arch/arm/mvp/commkm/comm.c deleted file mode 100644 index 8fd591c..0000000 --- a/arch/arm/mvp/commkm/comm.c +++ /dev/null @@ -1,1457 +0,0 @@ -/* - * Linux 2.6.32 and later Kernel module for VMware MVP Guest Communications - * - * Copyright (C) 2010-2012 VMware, Inc. All rights reserved. - * - * This program is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License version 2 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. - * - * You should have received a copy of the GNU General Public License along with - * this program; see the file COPYING. If not, write to the Free Software - * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. - */ -#line 5 - -/** - * @file - * - * @brief Communication functions based on transport functionality. - */ - -#include "comm.h" -#include "comm_transp_impl.h" - - -/* Constant and macro definitions */ - -#if defined(COMM_INSTRUMENT) -static CommOSAtomic commMaxCoalesceSize; -static CommOSAtomic commPacketsReceived; -static CommOSAtomic commCommittedPacketsReceived; -static CommOSAtomic commOpCalls; -#endif - -#define COMM_DISPATCH_EXTRA_WRITER_WAKEUP 1 - -#define COMM_CHANNEL_MAX_CAPACITY 2048 -#define COMM_CHANNEL_FREE 0x0 -#define COMM_CHANNEL_INITIALIZED 0x1 -#define COMM_CHANNEL_OPENED 0x2 -#define COMM_CHANNEL_ACTIVE 0x4 -#define COMM_CHANNEL_ZOMBIE 0x8 - -#define CommIsFree(chan) \ - ((chan)->lifecycleState == COMM_CHANNEL_FREE) -#define CommIsInitialized(chan) \ - ((chan)->lifecycleState == COMM_CHANNEL_INITIALIZED) -#define CommIsOpened(chan) \ - ((chan)->lifecycleState == COMM_CHANNEL_OPENED) -#define CommIsActive(chan) \ - ((chan)->lifecycleState == COMM_CHANNEL_ACTIVE) -#define CommIsZombie(chan) \ - ((chan)->lifecycleState == COMM_CHANNEL_ZOMBIE) - -#define CommSetFree(chan) \ - SetLifecycleState(chan, COMM_CHANNEL_FREE) -#define CommSetInitialized(chan) \ - SetLifecycleState(chan, COMM_CHANNEL_INITIALIZED) -#define CommSetOpened(chan) \ - SetLifecycleState(chan, COMM_CHANNEL_OPENED) -#define CommSetActive(chan) \ - SetLifecycleState(chan, COMM_CHANNEL_ACTIVE) -#define CommSetZombie(chan) \ - SetLifecycleState(chan, COMM_CHANNEL_ZOMBIE) - -#define CommGlobalLock() CommOS_SpinLock(&commGlobalLock) -#define CommGlobalUnlock() CommOS_SpinUnlock(&commGlobalLock) -#define CommGlobalLockBH() CommOS_SpinLockBH(&commGlobalLock) -#define CommGlobalUnlockBH() CommOS_SpinUnlockBH(&commGlobalLock) - -#define DispatchTrylock(chan) CommOS_MutexTrylock(&(chan)->dispatchMutex) -#define DispatchUnlock(chan) CommOS_MutexUnlock(&(chan)->dispatchMutex) - -#define WriteLock(chan) CommOS_MutexLock(&(chan)->writeMutex) -#define WriteTrylock(chan) CommOS_MutexTrylock(&(chan)->writeMutex) -#define WriteUnlock(chan) CommOS_MutexUnlock(&(chan)->writeMutex) - -#define StateLock(chan) CommOS_MutexLock(&(chan)->stateMutex) -#define StateTrylock(chan) CommOS_MutexTrylock(&(chan)->stateMutex) -#define StateUnlock(chan) CommOS_MutexUnlock(&(chan)->stateMutex) - -#define CommHoldInit(chan) CommOS_WriteAtomic(&(chan)->holds, 0) -#define CommHold(chan) CommOS_AddReturnAtomic(&(chan)->holds, 1) -#define CommRelease(chan) CommOS_SubReturnAtomic(&(chan)->holds, 1) -#define CommIsHeld(chan) (CommOS_ReadAtomic(&(chan)->holds) > 0) - -#define PacketLenOverLimit(chan, len) \ - (((len) - sizeof (CommPacket)) > ((chan)->transpArgs.capacity / 4)) - - -/* - * Data structure describing the offload <-> paravirtualized module - * communication channel. - */ - -struct CommChannelPriv { - CommOSAtomic holds; // Active readers and writers - CommTranspInitArgs transpArgs; // Transport initialization arguments - CommTransp transp; // Transport handle - CommOSMutex dispatchMutex; // Dispatch mutex - CommOSMutex writeMutex; // Non-BH write mutex - CommOSMutex stateMutex; // Upper-layer state mutex - CommOSWaitQueue availableWaitQ; // Available write space wait data - unsigned int desiredWriteSpace; // Size of write space needed - const CommImpl *impl; // Implementation - unsigned int implNmbOps; // Number of implementation operations - unsigned int lifecycleState; // Lifecycle state - void *state; // Upper layer-specific state -}; - - -static volatile int running; // Initialized and running. -static CommOSWaitQueue exitWaitQ; // Exit wait queue. -static CommOSSpinlock commGlobalLock; // Global lock. - - -/* Communication channel slots. */ - -static unsigned int commChannelCapacity; // Maximum number of channels. -static unsigned int commChannelSize; // Current size of channel array. -static unsigned int commChannelAllocated; // Nmb. entries currently in use. -static struct CommChannelPriv *commChannels; // Allocated channel array. - - -/** - * @brief Callback function called when the other side created a transport - * handle to which we need to potentially attach. - * @param[in,out] transpArgs arguments used when shared memory area was created. - * @param probeData our callback data, an implementation block. - * @return 0 if successful, -1 otherwise. - * @sideeffects May allocate a channel. - */ - -static int -DefaultTranspListener(CommTranspInitArgs *transpArgs, - void *probeData) -{ - int rc = -1; - const int inBH = 1; - const CommImpl *impl; - - if (!transpArgs || !probeData) { - CommOS_Debug(("%s: NULL args [0x%p, 0x%p].\n", - __FUNCTION__, transpArgs, probeData)); - goto out; - } - - impl = probeData; - CommOS_Debug(("%s: Received attach info [%u,%u,%u:%u].\n", - __FUNCTION__, - transpArgs->capacity, transpArgs->type, - transpArgs->id.d32[0], transpArgs->id.d32[1])); - - if (impl->checkArgs(transpArgs)) { - goto out; - } - transpArgs->mode = COMM_TRANSP_INIT_ATTACH; /* Ensure we attach. */ - - /* We recognized it, so don't let others waste any time. Even if we fail. */ - - rc = 0; - if (Comm_Alloc(transpArgs, impl, inBH, NULL)) { - impl->closeNtf(impl->closeNtfData, transpArgs, inBH); - CommOS_Log(("%s: Can't allocate new channel!\n", __FUNCTION__)); - } - -out: - return rc; -} - - -/** - * @brief Sets the lifecycle state of a channel entry - * @param channel channel to update - * @param newState state to update to - */ - -static inline void -SetLifecycleState(CommChannel channel, - unsigned int newState) -{ - - channel->lifecycleState = newState; -} - - -/* Wait conditions: functions returning 1: true, 0: false, < 0: error. */ - -/** - * @brief Wait condition function to check whether module can be unloaded. - * @param arg1 dummy - * @param arg2 dummy - * @return 1 if no channels are currently allocated, 0 if there are - */ - -static int -ExitCondition(void *arg1, - void *arg2) -{ - unsigned int i; - int rc; - - (void)arg1; - (void)arg2; - CommOS_Debug(("%s: running [%d] " - "commChannelAllocated [%u] commChannelSize [%u].\n", - __FUNCTION__, running, commChannelAllocated, commChannelSize)); - rc = !running && (commChannelAllocated == 0); - if (!rc) { - for (i = 0; i < commChannelCapacity; i++) { - CommOS_Debug(("%s: channel[%u] state [0x%x].\n", - __FUNCTION__, i, commChannels[i].lifecycleState)); - } - } - return rc; -} - - -/** - * @brief Wait condition function to check available write space. - * @param arg1 pointer to CommChannel struct - * @param arg2 size argument - * @return 1 if there is enough write space, 0 if not, -ENOMEM if comm down. - */ - -static int -WriteSpaceCondition(void *arg1, - void *arg2) -{ - CommChannel channel = arg1; - - if (!CommIsActive(channel)) { - return -ENOMEM; - } - return channel->desiredWriteSpace < CommTransp_EnqueueSpace(channel->transp); -} - - -/** - * @brief Registers an implementation block used when attaching to channels - * in response to transport attach events. - * @param impl implementation block. - * @return 0 if successful, non-zero otherwise. - */ - -int -Comm_RegisterImpl(const CommImpl *impl) -{ - CommTranspListener listener = { - .probe = DefaultTranspListener, - .probeData = (void *)impl - }; - - return CommTransp_Register(&listener); -} - - -/** - * @brief Unregisters an implementation block used when attaching to channels - * in response to transport attach events. - * @param impl implementation block. - */ - -void -Comm_UnregisterImpl(const CommImpl *impl) -{ - CommTranspListener listener = { - .probe = DefaultTranspListener, - .probeData = (void *)impl - }; - - CommTransp_Unregister(&listener); -} - - -/** - * @brief Allocates and initializes comm global state. Single-threaded use. - * @param maxChannels maximum number of channels. - * @return zero if successful, non-zero otherwise. - */ - -int -Comm_Init(unsigned int maxChannels) -{ - int rc = -1; - unsigned int i; - - if (running || commChannels || - (maxChannels == 0) || (maxChannels > COMM_CHANNEL_MAX_CAPACITY)) { - goto out; - } - -#if defined(COMM_INSTRUMENT) - CommOS_WriteAtomic(&commMaxCoalesceSize, 0); - CommOS_WriteAtomic(&commPacketsReceived, 0); - CommOS_WriteAtomic(&commCommittedPacketsReceived, 0); - CommOS_WriteAtomic(&commOpCalls, 0); -#endif - - CommOS_WaitQueueInit(&exitWaitQ); - CommOS_SpinlockInit(&commGlobalLock); - commChannelCapacity = maxChannels; - commChannelAllocated = 0; - commChannels = CommOS_Kmalloc((sizeof *commChannels) * commChannelCapacity); - if (!commChannels) { - goto out; - } - - memset(commChannels, 0, (sizeof *commChannels) * commChannelCapacity); - for (i = 0; i < commChannelCapacity; i++ ) { - CommChannel channel; - - channel = &commChannels[i]; - CommHoldInit(channel); - channel->transp = NULL; - CommOS_MutexInit(&channel->dispatchMutex); - CommOS_MutexInit(&channel->writeMutex); - CommOS_MutexInit(&channel->stateMutex); - CommOS_WaitQueueInit(&channel->availableWaitQ); - channel->desiredWriteSpace = -1U; - channel->state = NULL; - CommSetFree(channel); - } - - rc = CommTransp_Init(); - if (!rc) { - commChannelSize = 0; - running = 1; - rc = 0; - } else { - CommOS_Kfree(commChannels); - } - -out: - return rc; -} - - -/** - * @brief Initiates and finishes, comm global state deallocations. - * @param timeoutMillis initialization timeout in milliseconds - * @return zero if deallocations done, non-zero if more calls are needed. - */ - -int -Comm_Finish(unsigned long long *timeoutMillis) -{ - int rc; - unsigned int i; - unsigned long long timeout; - - for (i = 0; i < commChannelSize; i++) { - Comm_Zombify(&commChannels[i], 0); - } - - running = 0; - timeout = timeoutMillis ? *timeoutMillis : 0; - /* coverity[var_deref_model] */ - rc = CommOS_Wait(&exitWaitQ, ExitCondition, NULL, NULL, &timeout); - if (rc == 1) { - /* - * Didn't time out, task wasn't interrupted, we can wrap it up.. - */ - - CommTransp_Exit(); - CommOS_Kfree(commChannels); - commChannels = NULL; - commChannelSize = 0; -#if defined(COMM_INSTRUMENT) - CommOS_Log(("%s: commMaxCoalesceSize = %lu.\n", - __FUNCTION__, - CommOS_ReadAtomic(&commMaxCoalesceSize))); - CommOS_Log(("%s: commPacketsReceived = %lu.\n", - __FUNCTION__, - CommOS_ReadAtomic(&commPacketsReceived))); - CommOS_Log(("%s: commCommittedPacketsReceived = %lu.\n", - __FUNCTION__, - CommOS_ReadAtomic(&commCommittedPacketsReceived))); - CommOS_Log(("%s: commOpCalls = %lu.\n", - __FUNCTION__, - CommOS_ReadAtomic)(&commOpCalls))); -#endif - rc = 0; - } else { - rc = -1; - } - return rc; -} - - -/** - * @brief Finds a free entry and initializes it with the information provided. - * May be called from BH. It doesn't call potentially blocking functions. - * - * @note Depending on the choice of shared memory transport (VMCI or MVP QP), - * the 'inBH' distinction is important. VMCI datagrams are received under - * some circumstances in bottom-half context, so 'inBH' should be set. This - * is not a restriction on MVP. - * - * @param transpArgs transport initialization arguments. - * @param impl implementation block. - * @param inBH non-zero if called in bottom half. - * @param[out] newChannel newly allocated channel. - * @return zero if successful, non-zero otherwise. - * @sideeffects Initializes the communications channel with given parameters - */ - -int -Comm_Alloc(const CommTranspInitArgs *transpArgs, - const CommImpl *impl, - int inBH, - CommChannel *newChannel) -{ - unsigned int i; - CommChannel channel = NULL; - int restoreSize = 0; - int modHeld = 0; - int rc = -1; - - if (inBH) { - CommGlobalLock(); - } else { - CommGlobalLockBH(); - } - - if (!running || !transpArgs || !impl) { - goto out; - } - - if (CommOS_ModuleGet(impl->owner)) { - goto out; - } - modHeld = 1; - - for (i = 0; i < commChannelSize; i++) { - /* - * Check if this channel is already allocated. We don't match against - * ANY because those channels are in the process of being opened; after - * that happens, they'll get proper IDs. - */ - - if (!CommIsFree(&commChannels[i]) && - (transpArgs->id.d64 != COMM_TRANSP_ID_64_ANY) && - (transpArgs->id.d64 == commChannels[i].transpArgs.id.d64)) { - goto out; - } - if (!channel && CommIsFree(&commChannels[i])) { - channel = &commChannels[i]; - } - } - if (!channel) { - if (commChannelSize == commChannelCapacity) { - goto out; - } - channel = &commChannels[commChannelSize]; - commChannelSize++; - restoreSize = 1; - } - - if (channel->transp) { /* Inconsistency! */ - if (restoreSize) { - commChannelSize--; - } - goto out; - } - - channel->transpArgs = *transpArgs; - channel->impl = impl; - for (i = 0; impl->operations[i]; i++) { - ; - } - channel->implNmbOps = i; - channel->desiredWriteSpace = -1U; - commChannelAllocated++; - CommSetInitialized(channel); - if (newChannel) { - *newChannel = channel; - } - rc = 0; - CommOS_ScheduleDisp(); - -out: - if (inBH) { - CommGlobalUnlock(); - } else { - CommGlobalUnlockBH(); - } - if (rc && modHeld) { - CommOS_ModulePut(impl->owner); - } - return rc; -} - - -/** - * @brief Zombifies a channel. May fail if channel isn't active. - * @param[in,out] channel channel to zombify. - * @param inBH non-zero if called in bottom half. - * @return zero if channel zombified, non-zero otherwise. - */ - -int -Comm_Zombify(CommChannel channel, - int inBH) -{ - int rc = -1; - - if (!running) { - goto out; - } - if (inBH) { - CommGlobalLock(); - } else { - CommGlobalLockBH(); - } - if (CommIsActive(channel) || CommIsOpened(channel)) { - CommSetZombie(channel); - rc = 0; - } - if (inBH) { - CommGlobalUnlock(); - } else { - CommGlobalUnlockBH(); - } - -out: - if (!rc) { - CommOS_ScheduleDisp(); - } - return rc; -} - - -/** - * @brief Reports whether a channel is active. - * @param channel channel to report on. - * @return non-zero if channel active, zero otherwise. - */ - -int -Comm_IsActive(CommChannel channel) -{ - return channel ? CommIsActive(channel) : 0; -} - - -/** - * @brief Wakes up potential writer on the channel. This function must be - * called on an active channel, with either the dispatch lock taken, or - * the channel ref count incremented. - * @param channel CommChannel structure on which potential writer waits. - */ - -static inline void -WakeUpWriter(CommChannel channel) -{ - if (WriteSpaceCondition(channel, NULL)) { - CommOS_WakeUp(&channel->availableWaitQ); - } -} - - -/** - * @brief Transport event handler for comm channels. - * @param transp transport handle. - * @param event type of event. - * @param data callback data. - * @sideeffects may put the channel into zombie state, or schedule it for I/O. - */ - -static void -TranspEventHandler(CommTransp transp, - CommTranspIOEvent event, - void *data) -{ - CommChannel channel = (CommChannel)data; - - switch (event) { - case COMM_TRANSP_IO_DETACH: - CommOS_Debug(("%s: Detach event. Zombifying channel.\n", __FUNCTION__)); - Comm_Zombify(channel, 1); - break; - - case COMM_TRANSP_IO_IN: - case COMM_TRANSP_IO_INOUT: - /* - * The dispatch threads may not have been started because either: - * a) we're not running in the CommSvc service, or - * b) the Comm client didn't create them explicitly (CommOS_StartIO()). - * - * If so, the CommOS_ScheduleDisp() call is ineffective. This is - * the intended behavior: the client obviously wants to call the Comm - * dispatch function(s) directly. - */ - - CommOS_ScheduleDisp(); - break; - - case COMM_TRANSP_IO_OUT: - CommHold(channel); - if (CommIsActive(channel)) { - WakeUpWriter(channel); - } - CommRelease(channel); - if (CommIsZombie(channel)) { - /* - * After releasing the hold on the channel, we must check if it was - * set to zombie and the dispatcher was supposed to nuke it. If the - * dispatcher had made its run while we were holding the channel, it - * gave up. So schedule it. - */ - - CommOS_ScheduleDisp(); - } - break; - - default: - CommOS_Debug(("%s: Unhandled event [%u, %p, %p].\n", - __FUNCTION__, event, transp, data)); - } -} - - -/** - * @brief Destroys upper layer state, unregisters event handlers and - * detaches from or deletes shared memory. - * @param[in,out] channel CommChannel structure to close. - */ - -static void -CommClose(CommChannel channel) -{ - const CommImpl *impl = channel->impl; - - StateLock(channel); - if (impl->stateDtor && channel->state) { - impl->stateDtor(channel->state); - } - channel->state = NULL; - StateUnlock(channel); - - CommOS_ModulePut(impl->owner); - - if (channel->transp) { - CommTransp_Close(channel->transp); - channel->transp = NULL; - } - - CommGlobalLockBH(); - CommSetFree(channel); - commChannelAllocated--; - if (channel == &commChannels[commChannelSize - 1]) { - commChannelSize--; - } - CommGlobalUnlockBH(); - if (!running && (commChannelAllocated == 0)) { - CommOS_WakeUp(&exitWaitQ); - } -} - - -/** - * @brief Allocates upper layer state, registers transport event handler - * and creates or attaches to shared memory. - * @param[in,out] channel CommChannel structure to open. - * @return zero if successful, -1 otherwise - * @sideeffects Memory may be allocated, event handlers registered and - * QP allocated or attached to. - */ - -static int -CommOpen(CommChannel channel) -{ - int rc = -1; - CommTranspEvent transpEvent = { - .ioEvent = TranspEventHandler, - .ioEventData = channel - }; - const CommImpl *impl; - - if (!channel || !CommIsInitialized(channel)) { - return rc; - } - - if (!running) { /* Ok, toggle it back to FREE. */ - goto out; - } - - impl = channel->impl; - if (impl->stateCtor) { - channel->state = impl->stateCtor(channel); - if (!channel->state) { - goto out; - } - } - - if (!CommTransp_Open(&channel->transp, &channel->transpArgs, &transpEvent)) { - rc = 0; - } else { - channel->transp = NULL; - } - -out: - if (!rc) { - CommSetOpened(channel); - } else { - CommClose(channel); - } - return rc; -} - - -/** - * @brief Retrieves a channel's transport initialization arguments. - * It doesn't lock, the caller must ensure the channel may be accessed. - * @param channel CommChannel structure to get initialization arguments from. - * @return initialization arguments used to allocate/attach to channel. - */ - -CommTranspInitArgs -Comm_GetTranspInitArgs(CommChannel channel) -{ - if (!channel) { - CommTranspInitArgs res = { .capacity = 0 }; - - return res; - } - return channel->transpArgs; -} - - -/** - * @brief Retrieves upper layer state (pointer). It doesn't lock, the caller - * must ensure the channel may be accessed. - * @param channel CommChannel structure to get state from. - * @return pointer to upper layer state. - */ - -void * -Comm_GetState(CommChannel channel) -{ - if (!channel) { - return NULL; - } - return channel->state; -} - - -/** - * @brief Main input processing function operating on a given channel. - * @param channel CommChannel structure to process. - * @return number of processed channels (0 or 1), or -1 if channel closed. - * @sideeffects Lifecycle states are transitioned to and from. Channel may - * be opened or destroyed, waiting writers may be woken up, and input - * may be handed off to operation callbacks. - */ - -int -Comm_Dispatch(CommChannel channel) -{ - int rc = 0; - int zombify = 0; - CommPacket packet; - CommPacket firstPacket; - unsigned int dataLen; -#define VEC_SIZE 32 - struct kvec vec[VEC_SIZE]; - unsigned int vecLen; - - /* - * Taking the reader mutex is safe in all cases: entries, including - * free ones, are guaranteed to have initialized mutexes and locks. - * Locking empty entries may seem wasteful, but those entries are rare. - */ - - if (DispatchTrylock(channel)) { - return 0; - } - - /* Process input and writer wake-up. */ - - if (CommIsActive(channel)) { - /* - * The entry may have transitioned to ZOMBIE, somehow. That's OK - * since it can't be freed just yet (it's currently locked). - */ - - /* Wake up any waiting writers, if necessary. */ - - WakeUpWriter(channel); - - /* Read packets, payloads. */ - CommTransp_DequeueReset(channel->transp); - - for (vecLen = 0; vecLen < VEC_SIZE; vecLen++) { - if (!running) { - break; - } - - /* Read header. */ - - rc = CommTransp_DequeueSegment(channel->transp, - &packet, sizeof packet); - if (rc <= 0) { - /* No packet (header). */ - - rc = vecLen == 0 ? 0 : 1; - break; - } -#if defined(COMM_INSTRUMENT) - CommOS_AddReturnAtomic(commPacketsReceived, 1); -#endif - if ((rc != sizeof packet) || (packet.len < sizeof packet)) { - rc = -1; /* Fatal protocol error, close down comm. */ - break; - } - rc = 1; - - /* Read payload, if any. */ - - dataLen = packet.len - sizeof packet; - if (vecLen == 0) { - /* Save header of first packet. */ - - firstPacket = packet; - if (dataLen == 0) { - /* Commit no-payload packet read and we're done. */ - - CommTransp_DequeueCommit(channel->transp); -#if defined(COMM_INSTRUMENT) - CommOS_AddReturnAtomic(&commCommittedPacketsReceived, 1); -#endif - break; - } - } else { - /* - * Check if non-equivalent packet or above coalescing limit. - * If so, don't commit the read. - */ - - if (memcmp(&packet.opCode, &firstPacket.opCode, - sizeof packet - offsetof(CommPacket, opCode)) || - PacketLenOverLimit(channel, firstPacket.len + dataLen)) { - break; - } - } - - if (dataLen == 0) { - /* - * Received equivalent packet with zero-sized payload. This may - * happen in certain cases, such as pvtcp forwarding zero-sized - * datagrams. So don't break the loop, but keep going for as - * along as we can. - */ - - vec[vecLen].iov_base = NULL; - goto dequeueCommit; - } - - /* The packet has a payload (dataLen > 0). */ - - if (!(vec[vecLen].iov_base = channel->impl->dataAlloc(dataLen))) { - /* - * We treat out-of-(net?-)memory errors as "nothing to read". - * Memory pressure may either subside, in which case a future - * read may be successful, or be severe enough for the kernel - * to oops, anyway. Leave packet uncommitted. - */ - - CommOS_Debug(("%s: COULD NOT ALLOC PAYLOAD BYTES!\n", - __FUNCTION__)); - rc = vecLen == 0 ? 0 : 1; - break; - } - - /* Read payload and commit (packet and payload). */ - - rc = CommTransp_DequeueSegment(channel->transp, - vec[vecLen].iov_base, dataLen); - if (rc != dataLen) { - channel->impl->dataFree(vec[vecLen].iov_base); - CommOS_Log(("%s: BOOG -- COULD NOT DEQUEUE PAYLOAD! [%d != %u]", - __FUNCTION__, rc, dataLen)); - rc = -1; /* Fatal protocol error, close down comm. */ - break; - } - rc = 1; - -dequeueCommit: - CommTransp_DequeueCommit(channel->transp); -#if defined(COMM_INSTRUMENT) - CommOS_AddReturnAtomic(&commCommittedPacketsReceived, 1); -#endif - vec[vecLen].iov_len = dataLen; - if (vecLen > 0) { - firstPacket.len += dataLen; - if (packet.flags) { - /* Update to latest flags _iff_ latter non-zero. */ - - firstPacket.flags = packet.flags; - } - } -#if defined(COMM_INSTRUMENT) - if (firstPacket.len > - CommOS_ReadAtomic(&commMaxCoalesceSize)) { - CommOS_WriteAtomic(&commMaxCoalesceSize, firstPacket.len); - } -#endif - if (COMM_OPF_TEST_ERR(packet.flags)) { - /* If error bit is set, we're done (no more coalescing). */ - - vecLen++; - break; - } - } - - if (rc <= 0) { - if (rc < 0) { - zombify = 1; - rc = 1; - } - goto outUnlockAndFreeIovec; - } - -#if defined(COMM_DISPATCH_EXTRA_WRITER_WAKEUP) - /* Check again if we need to wake up any writers. */ - - WakeUpWriter(channel); -#endif - - if (firstPacket.opCode >= channel->implNmbOps) { - CommOS_Debug(("%s: Ignoring illegal opCode [%u]!\n", - __FUNCTION__, (unsigned int)firstPacket.opCode)); - CommOS_Debug(("%s: Max opCode: %u\n", - __FUNCTION__, channel->implNmbOps)); - goto outUnlockAndFreeIovec; - } - - /* - * NOTE: - * DispatchUnlock() _must_ be called from the operation callback. - * The reason for doing so is that, for better scalability, we want - * it released as soon as possible, BUT: - * - releasing it here, before calling into the operation, doesn't - * let the latter coordinate its own lock acquisition, such as - * potential socket or state locks. - * - alternatively, always releasing the dispatch lock after the - * operation completes, ties up the channel and imposes too much - * serialization between sockets. - * - to prevent the channel from being torn down while an operation - * is in flight (and potentially having released the dispatch lock), - * we increment the ref count on the channel and then release it - * after the function returns. - */ - -#if defined(COMM_INSTRUMENT) - CommOS_AddReturnAtomic(&commOpCalls, 1); -#endif - - CommHold(channel); - channel->impl->operations[firstPacket.opCode](channel, channel->state, - &firstPacket, vec, vecLen); - CommRelease(channel); - goto out; /* No unlocking, see comment above. */ - } - - /* Process state changes. */ - - if (CommIsZombie(channel) && !CommIsHeld(channel)) { - CommTranspInitArgs transpArgs = channel->transpArgs; - void (*closeNtf)(void *, - const CommTranspInitArgs *, - int inBH) = channel->impl->closeNtf; - void *closeNtfData = channel->impl->closeNtfData; - - while (WriteTrylock(channel)) { - /* Take the write lock; kick writers out if necessary. */ - - CommOS_Debug(("%s: Kicking writers out...\n", __FUNCTION__)); - CommOS_WakeUp(&channel->availableWaitQ); - } - WriteUnlock(channel); - - CommOS_Debug(("%s: Nuking zombie channel.\n", __FUNCTION__)); - CommClose(channel); - if (closeNtf) { - closeNtf(closeNtfData, &transpArgs, 0); - } - rc = -1; - } else if (CommIsInitialized(channel) && - (channel->impl->openAtMillis <= - CommOS_GetCurrentMillis())) { - if (!CommOpen(channel)) { - if (channel->transpArgs.mode == COMM_TRANSP_INIT_CREATE) { - /* - * If the attach side doesn't get notified, the entry will - * time out in OPENED and will be collected. - * Note that during the CommOpen(Transp_Open) call, the IDs - * in the transpArgs may have changed. Use those. - */ - - CommTransp_Notify(&channel->impl->ntfCenterID, - &channel->transpArgs); - } else { /* Attach mode */ - packet.len = sizeof packet; - packet.opCode = 0xff; - packet.flags = 0x00; - - /* - * Send out control packet, attach ack, and transition straight - * to ACTIVE. - */ - - rc = CommTransp_EnqueueAtomic(channel->transp, - &packet, sizeof packet); - if (rc == sizeof packet) { - /* Guard against potentially concurrent zombify. */ - - CommGlobalLockBH(); - if (CommIsOpened(channel)) { - CommOS_Debug(("%s: Sent attach ack. Activating channel.\n", - __FUNCTION__)); - CommSetActive(channel); - } - CommGlobalUnlockBH(); - } - } - rc = 1; - } - } else if (CommIsOpened(channel) && - (channel->transpArgs.mode == COMM_TRANSP_INIT_CREATE)) { - /* - * Get control packet (opCode == 0xff), attach ack (flags == 0x0), - * or check whether the channel timed out in OPENED. - */ - - rc = CommTransp_DequeueAtomic(channel->transp, - &packet, sizeof packet); - if (rc == sizeof packet) { - void (*activateNtf)(void *activateNtfData, CommChannel) = NULL; - void *activateNtfData = NULL; - - /* Guard against potentially concurrent zombify. */ - - CommGlobalLockBH(); - if (CommIsOpened(channel) && - (packet.opCode == 0xff) && (packet.flags == 0x0)) { - activateNtf = channel->impl->activateNtf; - activateNtfData = channel->impl->activateNtfData; - - CommSetActive(channel); - CommOS_Debug(("%s: Received attach ack. Activating channel.\n", - __FUNCTION__)); - } - CommHold(channel); - CommGlobalUnlockBH(); - - if (activateNtf) { - /* The callback must be short and 'put' the channel when done. */ - - activateNtf(activateNtfData, channel); - } else { - /* Don't forget to put back the channel if no activate callback. */ - - CommRelease(channel); - } - } else if ((channel->impl->openTimeoutAtMillis <= - CommOS_GetCurrentMillis()) || - !running) { - zombify = 1; - CommOS_Debug(("%s: Zombifying expired opened channel.\n", - __FUNCTION__)); - } - rc = 1; - } - DispatchUnlock(channel); - -out: - if (zombify) { - Comm_Zombify(channel, 0); - } - return rc; - -outUnlockAndFreeIovec: - DispatchUnlock(channel); - for ( ; vecLen; ) { - if (vec[--vecLen].iov_base) { - channel->impl->dataFree(vec[vecLen].iov_base); - vec[vecLen].iov_base = NULL; - } - vec[vecLen].iov_len = 0; - } - goto out; -#undef VEC_SIZE -} - - -/** - * @brief Main input processing function operating on all channels. - * @return number of processed channels. - * @sideeffects Lifecycle states are transitioned to and from. Channels may - * be opened and destroyed, waiting writers may be woken up, and input - * may be handed off to operation callbacks. - */ - -unsigned int -Comm_DispatchAll(void) -{ - unsigned int i; - unsigned int hits; - - for (hits = 0, i = 0; running && (i < commChannelSize); i++) { - hits += !!Comm_Dispatch(&commChannels[i]); - } - return hits; -} - - -/** - * @brief Writes a fully formatted packet (containing payload data, if - * applicable) to the specified channel. - * - * The operation may block until enough write space is available, but no - * more than the specified interval. The operation either writes the full - * amount of bytes, or it fails. Warning: callers must _not_ use the - * _Lock/_Unlock functions to bracket calls to this function. - * @param[in,out] channel channel to write to. - * @param packet packet to write. - * @param[in,out] timeoutMillis interval in milliseconds to wait. - * @return number of bytes written, 0 if it times out, -1 error. - * @sideeffects Data may be written to the channel. - */ - -int -Comm_Write(CommChannel channel, - const CommPacket *packet, - unsigned long long *timeoutMillis) -{ - int rc = -1; - int zombify; - - if (!channel || !timeoutMillis || - !packet || (packet->len < sizeof *packet)) { - return rc; - } - - zombify = (*timeoutMillis >= COMM_MAX_TO); - - WriteLock(channel); - if (!CommIsActive(channel)) { - goto out; - } - - CommTransp_EnqueueReset(channel->transp); - channel->desiredWriteSpace = packet->len; - rc = CommOS_DoWait(&channel->availableWaitQ, WriteSpaceCondition, - channel, NULL, timeoutMillis, - (*timeoutMillis != COMM_MAX_TO_UNINT)); - channel->desiredWriteSpace = -1U; - - if (rc) { /* Don't zombify, if it didn't time out. */ - zombify = 0; - } - if (rc == 1) { /* Enough write space, enqueue the packet. */ - rc = CommTransp_EnqueueAtomic(channel->transp, packet, packet->len); - if (rc != packet->len) { - zombify = 1; - rc = -1; /* Fatal protocol error. */ - } - } - -out: - WriteUnlock(channel); - if (zombify) { - Comm_Zombify(channel, 0); - } - return rc; -} - - -/** - * @brief Writes a packet and associated payload data to the specified channel. - * The operation may block until enough write space is available, but - * not more than the specified interval. - * The operation either writes the full amount of bytes, or it fails. - * If there is not enough data in the vector, padding will be added to - * reach the specified packet length, if the flags parameter requires it. - * Users may call this function successively to write several packets - * from large {io|k}vecs, when the flags parameter indicates it. If this - * is the case, the packet header needs to be updated accordingly in - * between calls, for the different (total) lengths. - * Warning: callers must _not_ use the _Lock/_Unlock functions to bracket - * calls to this function. - * @param[in,out] channel the specified channel. - * @param packet packet to write. - * @param[in,out] vec kvec to write from. - * @param[in,out] vecLen length of kvec. - * @param[in,out] timeoutMillis interval in milliseconds to wait. - * @param[in,out] iovOffset must be set to 0 before first call (internal cookie) - * @return number of bytes written, 0 if it timed out, -1 error. - * @sideeffects data may be written to the channel. - */ - -int -Comm_WriteVec(CommChannel channel, - const CommPacket *packet, - struct kvec **vec, - unsigned int *vecLen, - unsigned long long *timeoutMillis, - unsigned int *iovOffset) -{ - int rc; - int zombify; - unsigned int dataLen; - unsigned int vecDataLen; - unsigned int vecNdx; - unsigned int iovLen; - void *iovBase; - - if (!channel || !timeoutMillis || !iovOffset || - !packet || (packet->len < sizeof *packet) || - (((dataLen = packet->len - sizeof *packet) > 0) && - (!*vec || !*vecLen))) { - return -1; - } - - zombify = (*timeoutMillis >= COMM_MAX_TO); - - WriteLock(channel); - if (!CommIsActive(channel)) { - rc = -1; - goto out; - } - - CommTransp_EnqueueReset(channel->transp); - channel->desiredWriteSpace = packet->len; - rc = CommOS_DoWait(&channel->availableWaitQ, WriteSpaceCondition, - channel, NULL, timeoutMillis, - (*timeoutMillis != COMM_MAX_TO_UNINT)); - channel->desiredWriteSpace = -1U; - - if (rc) { /* Don't zombify, if it didn't time out. */ - zombify = 0; - } - if (rc == 1) { /* Enough write space, enqueue the packet. */ - iovLen = 0; - rc = CommTransp_EnqueueSegment(channel->transp, packet, sizeof *packet); - if (rc != sizeof *packet) { - zombify = 1; - rc = -1; /* Fatal protocol error. */ - goto out; - } - - if (dataLen > 0) { - int done = 0; - - for (vecDataLen = 0, vecNdx = 0; vecNdx < *vecLen; vecNdx++) { - if (vecNdx) { - *iovOffset = 0; - } - iovLen = (*vec)[vecNdx].iov_len - *iovOffset; - iovBase = (*vec)[vecNdx].iov_base + *iovOffset; - - if (!iovLen) { - continue; - } - - vecDataLen += iovLen; - if (vecDataLen >= dataLen) { - iovLen -= (vecDataLen - dataLen); - done = 1; - } - - rc = CommTransp_EnqueueSegment(channel->transp, iovBase, iovLen); - if (rc != iovLen) { - zombify = 1; - rc = -1; /* Fatal protocol error, close down comm. */ - goto out; - } - - if (done) { - CommTransp_EnqueueCommit(channel->transp); - if (vecDataLen == dataLen) { - vecNdx++; - *iovOffset = 0; - } else { - *iovOffset += iovLen; - } - *vecLen -= vecNdx; - *vec += vecNdx; - break; - } - } - - if (!done) { - /* - * We exhausted all the bytes in the given vector, but total length - * in the packet header is more than we sent (was available). - * If so, we pad by sending zero bytes to reach length required. - */ - - static char pad[1024]; - unsigned int delta; - unsigned int toSend; - - while (vecDataLen < dataLen) { - delta = dataLen - vecDataLen; - toSend = delta <= sizeof pad ? delta : sizeof pad; - if (toSend == delta) { - done = 1; - } - vecDataLen += toSend; - - rc = CommTransp_EnqueueSegment(channel->transp, pad, toSend); - if (rc != toSend) { - zombify = 1; - rc = -1; /* Fatal protocol error, close down comm. */ - goto out; - } - - if (done) { - CommTransp_EnqueueCommit(channel->transp); - *vec = NULL; - *vecLen = 0; - *iovOffset = 0; - break; - } - } - } - } else { - CommTransp_EnqueueCommit(channel->transp); - } - rc = (int)packet->len; - } else { - CommOS_Debug(("%s: timed out...\n", __FUNCTION__)); - } - -out: - WriteUnlock(channel); - if (zombify) { - Comm_Zombify(channel, 0); - } - return rc; -} - - -/** - * @brief Releases channel ref count. This function is exported for the upper - * layer's 'activateNtf' callback which may be run asynchronously. The - * callback is protected from concurrent channel releases until it calls - * this function. - * @param[in,out] channel CommChannel structure to release. - */ - -void -Comm_Put(CommChannel channel) -{ - if (channel) { - CommRelease(channel); - } -} - - -/** - * @brief Uses the read lock. This function is exported for the upper layer - * such that it can order acquisition of a different lock (socket) with - * the release of the dispatch lock. - * @param[in,out] channel CommChannel structure to unlock. - */ - -void -Comm_DispatchUnlock(CommChannel channel) -{ - if (channel) { - DispatchUnlock(channel); - } -} - - -/** - * @brief Lock the channel for upper layer state. - * This function is exported for the upper layer to ensure that channel - * isn't closed while updating the layer state. Operations using this - * function are expected to be short, since unlike the _Write functions, - * these callers cannot be signaled. - * @param[in,out] channel CommChannel structure to lock. - * @return zero if successful, -1 otherwise. - */ - -int -Comm_Lock(CommChannel channel) -{ - if (!channel) { - return -1; - } - StateLock(channel); - if (!CommIsActive(channel) && !CommIsZombie(channel)) { - StateUnlock(channel); - return -1; - } - return 0; -} - - -/** - * @brief Uses the writer lock. This function is exported for the upper layer - * to ensure that channel isn't closed while updating the layer state. - * See Comm_Lock for details). - * @param[in,out] channel CommChannel structure to unlock. - */ - -void -Comm_Unlock(CommChannel channel) -{ - if (channel) { - StateUnlock(channel); - } -} - - -/** - * @brief Requests events be posted in-line after the function completes. - * @param channel channel object. - * @return current number of requests for inline event posting, or -1 on error. - */ - -unsigned int -Comm_RequestInlineEvents(CommChannel channel) -{ - if (channel->transp) { - return CommTransp_RequestInlineEvents(channel->transp); - } else { - return (unsigned int)-1; - } -} - - -/** - * @brief Requests events be posted out-of-band after the function completes. - * @param channel channel object. - * @return current number of requests for inline event posting, or -1 on error. - */ - -unsigned int -Comm_ReleaseInlineEvents(CommChannel channel) -{ - if (channel->transp) { - return CommTransp_ReleaseInlineEvents(channel->transp); - } else { - return (unsigned int)-1; - } -} |