zephyr/kernel/microkernel/k_pipe_xfer.c

1064 lines
28 KiB
C

/* pipe processing for data transfer */
/*
* Copyright (c) 1997-2015 Wind River Systems, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <micro_private.h>
#include <k_pipe_buffer.h>
#include <k_pipe_util.h>
#include <toolchain.h>
#include <sections.h>
#include <misc/__assert.h>
#include <misc/util.h>
#define FORCE_XFER_ON_STALL
#define _X_TO_N (_0_TO_N | _1_TO_N)
/*
* - artefacts: ???
* - non-optimal:
* from single requester to multiple requesters : basic function is
* pipe_read_write() - copies remaining data into buffer; better would be to
* possibly copy the remaining data to the next requester (if there is one)
*/
/**
*
* _k_pipe_movedata_ack -
*
* @return N/A
*/
void _k_pipe_movedata_ack(struct k_args *pEOXfer)
{
struct _pipe_xfer_ack_arg *pipe_xfer_ack = &pEOXfer->args.pipe_xfer_ack;
switch (pipe_xfer_ack->xfer_type) {
case XFER_W2B: /* Writer to Buffer */
{
struct k_args *writer_ptr = pipe_xfer_ack->writer_ptr;
if (writer_ptr) { /* Xfer from Writer finished */
struct _pipe_xfer_req_arg *pipe_write_req =
&pipe_xfer_ack->writer_ptr->args.pipe_xfer_req;
--pipe_write_req->num_pending_xfers;
if (pipe_write_req->num_pending_xfers == 0) {
if (TERM_XXX & pipe_write_req->status) {
/* request is terminated, send reply */
_k_pipe_put_reply(pipe_xfer_ack->writer_ptr);
/* invoke continuation mechanism (fall through) */
} else {
/* invoke continuation mechanism (fall through) */
}
} else {
if (TERM_XXX & pipe_write_req->status) {
/* do nothing */
/* invoke continuation mechanism (fall through) */
} else {
/* invoke continuation mechanism (fall through) */
}
}
} else {
/* Xfer to Buffer finished */
int XferId = pipe_xfer_ack->id;
BuffEnQA_End(&pipe_xfer_ack->pipe_ptr->desc, XferId,
pipe_xfer_ack->size);
}
/* invoke continuation mechanism */
_k_pipe_process(pipe_xfer_ack->pipe_ptr, NULL, NULL);
FREEARGS(pEOXfer);
return;
} /* XFER_W2B */
case XFER_B2R: {
struct k_args *reader_ptr = pipe_xfer_ack->reader_ptr;
if (reader_ptr) { /* Xfer to Reader finished */
struct _pipe_xfer_req_arg *pipe_read_req =
&pipe_xfer_ack->reader_ptr->args.pipe_xfer_req;
--pipe_read_req->num_pending_xfers;
if (pipe_read_req->num_pending_xfers == 0) {
if (TERM_XXX & pipe_read_req->status) {
/* request is terminated, send reply */
_k_pipe_get_reply(pipe_xfer_ack->reader_ptr);
} else {
/* invoke continuation mechanism (fall through) */
}
} else {
if (TERM_XXX & pipe_read_req->status) {
/* do nothing */
/* invoke continuation mechanism (fall through) */
} else {
/* invoke continuation mechanism (fall through) */
}
}
} else {
/* Xfer from Buffer finished */
int XferId = pipe_xfer_ack->id;
BuffDeQA_End(&pipe_xfer_ack->pipe_ptr->desc, XferId,
pipe_xfer_ack->size);
}
/* continuation mechanism */
_k_pipe_process(pipe_xfer_ack->pipe_ptr, NULL, NULL);
FREEARGS(pEOXfer);
return;
} /* XFER_B2R */
case XFER_W2R: {
struct k_args *writer_ptr = pipe_xfer_ack->writer_ptr;
if (writer_ptr) { /* Transfer from writer finished */
struct _pipe_xfer_req_arg *pipe_write_req =
&pipe_xfer_ack->writer_ptr->args.pipe_xfer_req;
--pipe_write_req->num_pending_xfers;
if (pipe_write_req->num_pending_xfers == 0) {
if (TERM_XXX & pipe_write_req->status) {
/* request is terminated, send reply */
_k_pipe_put_reply(pipe_xfer_ack->writer_ptr);
} else {
/* invoke continuation mechanism (fall through) */
}
} else {
if (TERM_XXX & pipe_write_req->status) {
/* do nothing */
/* invoke continuation mechanism (fall through) */
} else {
/* invoke continuation mechanism (fall through) */
}
}
} else {
/* Transfer to Reader finished */
struct _pipe_xfer_req_arg *pipe_read_req =
&pipe_xfer_ack->reader_ptr->args.pipe_xfer_req;
--pipe_read_req->num_pending_xfers;
if (pipe_read_req->num_pending_xfers == 0) {
if (TERM_XXX & pipe_read_req->status) {
/* request is terminated, send reply */
_k_pipe_get_reply(pipe_xfer_ack->reader_ptr);
} else {
/* invoke continuation mechanism (fall through) */
}
} else {
if (TERM_XXX & pipe_read_req->status) {
/* do nothing */
/* invoke continuation mechanism (fall through) */
} else {
/* invoke continuation mechanism (fall through) */
}
}
}
/* invoke continuation mechanism */
_k_pipe_process(pipe_xfer_ack->pipe_ptr, NULL, NULL);
FREEARGS(pEOXfer);
return;
} /* XFER_W2B */
default:
break;
}
}
/**
*
* @brief Determines priority for data move operation
*
* Uses priority level of most important participant.
*
* Note: It's OK to have one or two participants, but there can't be none!
*
* @return N/A
*/
static kpriority_t move_priority_compute(struct k_args *writer_ptr,
struct k_args *reader_ptr)
{
kpriority_t move_priority;
if (!writer_ptr) {
move_priority = reader_ptr->priority;
} else {
move_priority = writer_ptr->priority;
if (reader_ptr && (reader_ptr->priority < move_priority)) {
move_priority = reader_ptr->priority;
}
}
return move_priority;
}
/**
*
* setup_movedata -
*
* @return N/A
*/
static void setup_movedata(struct k_args *A,
struct _k_pipe_struct *pipe_ptr, XFER_TYPE xfer_type,
struct k_args *writer_ptr, struct k_args *reader_ptr,
void *destination, void *source,
uint32_t size, int XferID)
{
struct k_args *pContSend;
struct k_args *pContRecv;
A->Comm = _K_SVC_MOVEDATA_REQ;
A->Ctxt.task = NULL;
/*
* this caused problems when != NULL related to set/reset of
* state bits
*/
A->args.moved_req.action = (MovedAction)(MVDACT_SNDACK | MVDACT_RCVACK);
A->args.moved_req.source = source;
A->args.moved_req.destination = destination;
A->args.moved_req.total_size = size;
/* continuation packet */
GETARGS(pContSend);
GETARGS(pContRecv);
pContSend->next = NULL;
pContSend->Comm = _K_SVC_PIPE_MOVEDATA_ACK;
pContSend->args.pipe_xfer_ack.pipe_ptr = pipe_ptr;
pContSend->args.pipe_xfer_ack.xfer_type = xfer_type;
pContSend->args.pipe_xfer_ack.id = XferID;
pContSend->args.pipe_xfer_ack.size = size;
pContRecv->next = NULL;
pContRecv->Comm = _K_SVC_PIPE_MOVEDATA_ACK;
pContRecv->args.pipe_xfer_ack.pipe_ptr = pipe_ptr;
pContRecv->args.pipe_xfer_ack.xfer_type = xfer_type;
pContRecv->args.pipe_xfer_ack.id = XferID;
pContRecv->args.pipe_xfer_ack.size = size;
A->priority = move_priority_compute(writer_ptr, reader_ptr);
pContSend->priority = A->priority;
pContRecv->priority = A->priority;
switch (xfer_type) {
case XFER_W2B: /* Writer to Buffer */
{
__ASSERT_NO_MSG(reader_ptr == NULL);
pContSend->args.pipe_xfer_ack.writer_ptr = writer_ptr;
pContRecv->args.pipe_xfer_ack.writer_ptr = NULL;
break;
}
case XFER_B2R: {
__ASSERT_NO_MSG(writer_ptr == NULL);
pContSend->args.pipe_xfer_ack.reader_ptr = NULL;
pContRecv->args.pipe_xfer_ack.reader_ptr = reader_ptr;
break;
}
case XFER_W2R: {
__ASSERT_NO_MSG(writer_ptr != NULL && reader_ptr != NULL);
pContSend->args.pipe_xfer_ack.writer_ptr = writer_ptr;
pContSend->args.pipe_xfer_ack.reader_ptr = NULL;
pContRecv->args.pipe_xfer_ack.writer_ptr = NULL;
pContRecv->args.pipe_xfer_ack.reader_ptr = reader_ptr;
break;
}
default:
__ASSERT_NO_MSG(1 == 0); /* we should not come here */
}
A->args.moved_req.extra.setup.continuation_send = pContSend;
A->args.moved_req.extra.setup.continuation_receive = pContRecv;
/*
* (possible optimisation)
* if we could know if it was a send/recv completion, we could use the
* SAME cmd packet for continuation on both completion of send and recv !!
*/
}
static int ReaderInProgressIsBlocked(struct _k_pipe_struct *pipe_ptr,
struct k_args *reader_ptr)
{
int iSizeSpaceInReader;
int iAvailBufferData;
TIME_TYPE TimeType;
K_PIPE_OPTION option;
/* first condition: request cannot wait any longer: must be -
* (non-blocked) or a finite timed wait with a killed timer
*/
TimeType = _k_pipe_time_type_get(&reader_ptr->args);
option = _k_pipe_option_get(&reader_ptr->args);
if (((TimeType == _TIME_B) && (option == _ALL_N)) ||
((TimeType == _TIME_B) && (_X_TO_N & option) &&
!(reader_ptr->args.pipe_xfer_req.xferred_size))
#ifdef CANCEL_TIMERS
|| ((TimeType == _TIME_BT) && reader_ptr->Time.timer)
#endif
) {
/*
* requester can still wait (for some time or forever), no
* problem for now
*/
return 0;
}
/* second condition: buffer activity is null */
if (pipe_ptr->desc.num_pending_writes != 0 ||
pipe_ptr->desc.num_pending_reads != 0) {
/*
* buffer activity detected, can't say now that processing is
* blocked
*/
return 0;
}
/* third condition: */
iSizeSpaceInReader =
reader_ptr->args.pipe_xfer_req.total_size -
reader_ptr->args.pipe_xfer_req.xferred_size;
BuffGetAvailDataTotal(&pipe_ptr->desc, &iAvailBufferData);
if (iAvailBufferData >= iSizeSpaceInReader) {
return 0;
} else {
return 1;
}
}
static int WriterInProgressIsBlocked(struct _k_pipe_struct *pipe_ptr,
struct k_args *writer_ptr)
{
int iSizeDataInWriter;
int iFreeBufferSpace;
TIME_TYPE TimeType;
K_PIPE_OPTION option;
/*
* first condition: request cannot wait any longer: must be -
* (non-blocked) or a finite timed wait with a killed timer
*/
TimeType = _k_pipe_time_type_get(&writer_ptr->args);
option = _k_pipe_option_get(&writer_ptr->args);
if (((TimeType == _TIME_B) && (option == _ALL_N)) ||
((TimeType == _TIME_B) && (_X_TO_N & option) &&
!(writer_ptr->args.pipe_xfer_req.xferred_size))
#ifdef CANCEL_TIMERS
|| ((TimeType == _TIME_BT) && writer_ptr->Time.timer)
#endif
) {
/*
* requester can still wait (for some time or forever), no
* problem for now
*/
return 0;
}
/* second condition: buffer activity is null */
if (pipe_ptr->desc.num_pending_writes != 0 ||
pipe_ptr->desc.num_pending_reads != 0) {
/*
* buffer activity detected, can't say now that processing is
* blocked
*/
return 0;
}
/* third condition: */
iSizeDataInWriter =
writer_ptr->args.pipe_xfer_req.total_size -
writer_ptr->args.pipe_xfer_req.xferred_size;
BuffGetFreeSpaceTotal(&pipe_ptr->desc, &iFreeBufferSpace);
if (iFreeBufferSpace >= iSizeDataInWriter) {
return 0;
} else {
return 1;
}
}
/**
*
* @brief Read from the pipe
*
* This routine reads from the pipe. If <pipe_ptr> is NULL, then it uses
* <pNewReader> as the reader. Otherwise it takes the reader from the pipe
* structure.
*
* @return N/A
*/
static void pipe_read(struct _k_pipe_struct *pipe_ptr,
struct k_args *pNewReader)
{
struct k_args *reader_ptr;
struct _pipe_xfer_req_arg *pipe_read_req;
unsigned char *read_ptr;
int size;
int id;
int ret;
int numIterations = 2;
reader_ptr = (pNewReader != NULL) ? pNewReader : pipe_ptr->readers;
__ASSERT_NO_MSG((pipe_ptr->readers == pNewReader) ||
(pipe_ptr->readers == NULL) || (pNewReader == NULL));
pipe_read_req = &reader_ptr->args.pipe_xfer_req;
do {
size = min(pipe_ptr->desc.available_data_count,
pipe_read_req->total_size - pipe_read_req->xferred_size);
if (size == 0) {
return;
}
struct k_args *Moved_req;
ret = BuffDeQA(&pipe_ptr->desc, size, &read_ptr, &id);
if (ret == 0) {
return;
}
GETARGS(Moved_req);
setup_movedata(Moved_req, pipe_ptr, XFER_B2R, NULL, reader_ptr,
(char *)(pipe_read_req->data_ptr) +
OCTET_TO_SIZEOFUNIT(pipe_read_req->xferred_size),
read_ptr, ret, id);
_k_movedata_request(Moved_req);
FREEARGS(Moved_req);
pipe_read_req->num_pending_xfers++;
pipe_read_req->xferred_size += ret;
if (pipe_read_req->xferred_size == pipe_read_req->total_size) {
_k_pipe_request_status_set(pipe_read_req,
TERM_SATISFIED);
if (reader_ptr->head != NULL) {
DeListWaiter(reader_ptr);
myfreetimer(&reader_ptr->Time.timer);
}
return;
}
_k_pipe_request_status_set(pipe_read_req, XFER_BUSY);
} while (--numIterations != 0);
}
/**
*
* @brief Write to the pipe
*
* This routine writes to the pipe. If <pipe_ptr> is NULL, then it uses
* <pNewWriter> as the writer. Otherwise it takes the writer from the pipe
* structure.
*
* @return N/A
*/
static void pipe_write(struct _k_pipe_struct *pipe_ptr,
struct k_args *pNewWriter)
{
struct k_args *writer_ptr;
struct _pipe_xfer_req_arg *pipe_write_req;
int size;
unsigned char *write_ptr;
int id;
int ret;
int numIterations = 2;
writer_ptr = (pNewWriter != NULL) ? pNewWriter : pipe_ptr->writers;
__ASSERT_NO_MSG(!((pipe_ptr->writers != pNewWriter) &&
(pipe_ptr->writers != NULL) && (pNewWriter != NULL)));
pipe_write_req = &writer_ptr->args.pipe_xfer_req;
do {
size = min((numIterations == 2) ? pipe_ptr->desc.free_space_count
: pipe_ptr->desc.free_space_post_wrap_around,
pipe_write_req->total_size - pipe_write_req->xferred_size);
if (size == 0) {
continue;
}
struct k_args *Moved_req;
ret = BuffEnQA(&pipe_ptr->desc, size, &write_ptr, &id);
if (ret == 0) {
return;
}
GETARGS(Moved_req);
setup_movedata(Moved_req, pipe_ptr, XFER_W2B, writer_ptr, NULL,
write_ptr, (char *)(pipe_write_req->data_ptr) +
OCTET_TO_SIZEOFUNIT(pipe_write_req->xferred_size),
ret, (numIterations == 2) ? id : -1);
_k_movedata_request(Moved_req);
FREEARGS(Moved_req);
pipe_write_req->num_pending_xfers++;
pipe_write_req->xferred_size += ret;
if (pipe_write_req->xferred_size == pipe_write_req->total_size) {
_k_pipe_request_status_set(pipe_write_req,
TERM_SATISFIED);
if (writer_ptr->head != NULL) {
/* only listed requests have a timer */
DeListWaiter(writer_ptr);
myfreetimer(&writer_ptr->Time.timer);
}
return;
}
_k_pipe_request_status_set(pipe_write_req, XFER_BUSY);
} while (--numIterations != 0);
}
/**
* @brief Update the pipe transfer status
*
* @param pActor pointer to struct k_args to be used by actor
* @param pipe_xfer_req pointer to actor's pipe process structure
* @param bytesXferred number of bytes transferred
*
* @return N/A
*/
static void pipe_xfer_status_update(struct k_args *pActor,
struct _pipe_xfer_req_arg *pipe_xfer_req,
int bytesXferred)
{
pipe_xfer_req->num_pending_xfers++;
pipe_xfer_req->xferred_size += bytesXferred;
if (pipe_xfer_req->xferred_size == pipe_xfer_req->total_size) {
_k_pipe_request_status_set(pipe_xfer_req, TERM_SATISFIED);
if (pActor->head != NULL) {
DeListWaiter(pActor);
myfreetimer(&pActor->Time.timer);
}
} else {
_k_pipe_request_status_set(pipe_xfer_req, XFER_BUSY);
}
}
/**
* @brief Read and/or write from/to the pipe
*
* @param pipe_ptr pointer to pipe structure
* @param pNewWriter pointer to new writer struct k_args
* @param pNewReader pointer to new reader struct k_args
*
* @return N/A
*/
static void pipe_read_write(struct _k_pipe_struct *pipe_ptr,
struct k_args *pNewWriter,
struct k_args *pNewReader)
{
/* ptr to struct k_args to be used by reader */
struct k_args *reader_ptr;
/* ptr to struct k_args to be used by writer */
struct k_args *writer_ptr;
/* ptr to writer's pipe process structure */
struct _pipe_xfer_req_arg *pipe_write_req;
/* ptr to reader's pipe process structure */
struct _pipe_xfer_req_arg *pipe_read_req;
int iT1;
int iT2;
int iT3;
writer_ptr = (pNewWriter != NULL) ? pNewWriter : pipe_ptr->writers;
__ASSERT_NO_MSG((pipe_ptr->writers == pNewWriter) ||
(pipe_ptr->writers == NULL) || (pNewWriter == NULL));
reader_ptr = (pNewReader != NULL) ? pNewReader : pipe_ptr->readers;
__ASSERT_NO_MSG((pipe_ptr->readers == pNewReader) ||
(pipe_ptr->readers == NULL) || (pNewReader == NULL));
/* Preparation */
pipe_write_req = &writer_ptr->args.pipe_xfer_req;
pipe_read_req = &reader_ptr->args.pipe_xfer_req;
/* Calculate iT1, iT2 and iT3 */
int iFreeSpaceReader = (pipe_read_req->total_size -
pipe_read_req->xferred_size);
int iAvailDataWriter = (pipe_write_req->total_size -
pipe_write_req->xferred_size);
int iFreeSpaceBuffer = (pipe_ptr->desc.free_space_count +
pipe_ptr->desc.free_space_post_wrap_around);
int iAvailDataBuffer = (pipe_ptr->desc.available_data_count +
pipe_ptr->desc.available_data_post_wrap_around);
iT1 = min(iFreeSpaceReader, iAvailDataBuffer);
iFreeSpaceReader -= iT1;
if (pipe_ptr->desc.num_pending_writes == 0) {
/* no incoming data anymore */
iT2 = min(iFreeSpaceReader, iAvailDataWriter);
iAvailDataWriter -= iT2;
iT3 = min(iAvailDataWriter, iFreeSpaceBuffer);
} else {
/*
* There is still data coming into the buffer from a writer.
* Therefore <iT2> must be zero; there is no direct W-to-R
* transfer as the buffer is not really 'empty'.
*/
iT2 = 0;
iT3 = 0; /* this is a choice (can be optimised later on) */
}
/***************/
/* ACTION !!!! */
/***************/
/* T1 transfer */
if (iT1 != 0) {
pipe_read(pipe_ptr, reader_ptr);
}
/* T2 transfer */
if (iT2 != 0) {
struct k_args *Moved_req;
__ASSERT_NO_MSG(reader_ptr->args.pipe_xfer_req.status != TERM_SATISFIED);
GETARGS(Moved_req);
setup_movedata(Moved_req, pipe_ptr, XFER_W2R,
writer_ptr, reader_ptr,
(char *)(pipe_read_req->data_ptr) +
OCTET_TO_SIZEOFUNIT(pipe_read_req->xferred_size),
(char *)(pipe_write_req->data_ptr) +
OCTET_TO_SIZEOFUNIT(pipe_write_req->xferred_size),
iT2, -1);
_k_movedata_request(Moved_req);
FREEARGS(Moved_req);
pipe_xfer_status_update(writer_ptr, pipe_write_req, iT2);
pipe_xfer_status_update(reader_ptr, pipe_read_req, iT2);
}
/* T3 transfer */
if (iT3 != 0) {
__ASSERT_NO_MSG(TERM_SATISFIED != writer_ptr->args.pipe_xfer_req.status);
pipe_write(pipe_ptr, writer_ptr);
}
}
void _k_pipe_process(struct _k_pipe_struct *pipe_ptr, struct k_args *pNLWriter,
struct k_args *pNLReader)
{
struct k_args *reader_ptr = NULL;
struct k_args *writer_ptr = NULL;
__ASSERT_NO_MSG(!(pNLWriter && pNLReader));
/* both a pNLWriter and pNLReader, is that allowed?
* Code below has not been designed for that.
* Anyway, this will not happen in current version.
*/
struct k_args *pNextReader;
struct k_args *pNextWriter;
do {
bool bALLNWriterNoGo = false;
bool bALLNReaderNoGo = false;
/* Reader */
if (pNLReader != NULL) {
if (reader_ptr != pNLReader) {
pNextReader = pipe_ptr->readers;
if (pNextReader == NULL) {
if (!(TERM_XXX & pNLReader->args.pipe_xfer_req.status))
pNextReader = pNLReader;
}
} else {
/* we already used the extra non-listed Reader */
if (TERM_XXX & reader_ptr->args.pipe_xfer_req.status) {
pNextReader = NULL;
} else {
/* == pNLReader */
pNextReader = reader_ptr;
}
}
} else {
pNextReader = pipe_ptr->readers;
}
/* Writer */
if (pNLWriter != NULL) {
if (writer_ptr != pNLWriter) {
pNextWriter = pipe_ptr->writers;
if (pNextWriter == NULL) {
if (!(TERM_XXX & pNLWriter->args.pipe_xfer_req.status))
pNextWriter = pNLWriter;
}
} else {
/* we already used the extra non-listed Writer */
if (TERM_XXX & writer_ptr->args.pipe_xfer_req.status) {
pNextWriter = NULL;
} else {
pNextWriter = writer_ptr;
}
}
} else {
pNextWriter = pipe_ptr->writers;
}
/* check if there is uberhaupt something to do */
if (pNextReader == NULL && pNextWriter == NULL)
return;
if (pNextReader == reader_ptr && pNextWriter == writer_ptr)
break; /* nothing changed, so stop */
/* go with pNextReader and pNextWriter */
reader_ptr = pNextReader;
writer_ptr = pNextWriter;
if (writer_ptr) {
if (_k_pipe_option_get(&writer_ptr->args) == _ALL_N &&
(writer_ptr->args.pipe_xfer_req.xferred_size == 0) &&
_k_pipe_time_type_get(&writer_ptr->args) != _TIME_B) {
/* investigate if there is a problem for
* his request to be satisfied
*/
int iSizeDataInWriter;
int iSpace2WriteinReaders;
int iFreeBufferSpace;
int iTotalSpace2Write;
iSpace2WriteinReaders =
CalcFreeReaderSpace(pipe_ptr->readers);
if (pNLReader)
iSpace2WriteinReaders +=
(pNLReader->args.pipe_xfer_req.total_size -
pNLReader->args.pipe_xfer_req.xferred_size);
BuffGetFreeSpaceTotal(&pipe_ptr->desc, &iFreeBufferSpace);
iTotalSpace2Write =
iFreeBufferSpace + iSpace2WriteinReaders;
iSizeDataInWriter =
writer_ptr->args.pipe_xfer_req.total_size -
writer_ptr->args.pipe_xfer_req.xferred_size;
if (iSizeDataInWriter > iTotalSpace2Write) {
bALLNWriterNoGo = true;
}
}
}
if (reader_ptr) {
if (_k_pipe_option_get(&reader_ptr->args) == _ALL_N &&
(reader_ptr->args.pipe_xfer_req.xferred_size == 0) &&
_k_pipe_time_type_get(&reader_ptr->args) != _TIME_B) {
/* investigate if there is a problem for
* his request to be satisfied
*/
int iSizeFreeSpaceInReader;
int iData2ReadFromWriters;
int iAvailBufferData;
int iTotalData2Read;
iData2ReadFromWriters = CalcAvailWriterData(pipe_ptr->writers);
if (pNLWriter)
iData2ReadFromWriters +=
(pNLWriter->args.pipe_xfer_req.total_size -
pNLWriter->args.pipe_xfer_req.xferred_size);
BuffGetAvailDataTotal(&pipe_ptr->desc, &iAvailBufferData);
iTotalData2Read = iAvailBufferData + iData2ReadFromWriters;
iSizeFreeSpaceInReader =
reader_ptr->args.pipe_xfer_req.total_size -
reader_ptr->args.pipe_xfer_req.xferred_size;
if (iSizeFreeSpaceInReader > iTotalData2Read) {
bALLNReaderNoGo = true;
}
}
}
__ASSERT_NO_MSG(!(bALLNWriterNoGo && bALLNReaderNoGo));
/************/
/* ACTION: */
/************/
if (bALLNWriterNoGo) {
/* investigate if we must force a transfer to avoid a stall */
if (!BuffEmpty(&pipe_ptr->desc)) {
if (reader_ptr) {
pipe_read(pipe_ptr, reader_ptr);
continue;
} else {
/* we could break as well,
* but then nothing else will happen
*/
return;
}
} else {
#ifdef FORCE_XFER_ON_STALL
if (reader_ptr &&
(_k_pipe_time_type_get(&writer_ptr->args) != _TIME_NB)) {
/* force transfer (we make exception
* for non-blocked writer)
*/
pipe_read_write(pipe_ptr, writer_ptr, reader_ptr);
continue;
} else
#endif
/* we could break as well,
* but then nothing else will happen
*/
return;
}
} else if (bALLNReaderNoGo) {
/*
* investigate if we must force a transfer to avoid a
* stall
*/
if (!BuffFull(&pipe_ptr->desc)) {
if (writer_ptr) {
pipe_write(pipe_ptr, writer_ptr);
continue;
} else {
return;
}
} else {
#ifdef FORCE_XFER_ON_STALL
if (writer_ptr &&
(_k_pipe_time_type_get(&reader_ptr->args) != _TIME_NB)) {
/* force transfer (we make exception
* for non-blocked reader)
*/
pipe_read_write(pipe_ptr, writer_ptr, reader_ptr);
continue;
} else
#endif
return;
}
} else {
/* no blocked reader and no blocked writer (if there
* are any of them) == NOMINAL operation
*/
if (reader_ptr) {
if (writer_ptr) {
pipe_read_write(pipe_ptr, writer_ptr,
reader_ptr);
continue;
} else {
pipe_read(pipe_ptr, reader_ptr);
continue;
}
} else {
if (writer_ptr) {
pipe_write(pipe_ptr, writer_ptr);
continue;
} else {
/* we should not come here */
__ASSERT_NO_MSG(1 == 0);
return;
}
}
}
} while (1);
/* We stopped processing because nothing changed anymore (stall)
* Let's examine the situation a little bit further
*/
reader_ptr = pNextReader;
writer_ptr = pNextWriter;
/* if we come here, it is b/c reader_ptr and writer_ptr did not change
* anymore.
* - Normally one of them is NULL, which means only a writer, resp. a
* reader remained.
* - The case that none of them is NULL is a special case which
* 'normally' does not occur.
* A remaining reader_ptr and/or writer_ptr are expected to be
* not-completed.
*
* Note that in the case there is only a reader or there is only a
* writer, it can be a ALL_N request.
* This happens when his request has not been processed completely yet
* (b/c of a copy in and copy out conflict in the buffer e.g.), but is
* expected to be processed completely somewhat later (must be !)
*/
/* in the sequel, we will:
* 1. check the hypothesis that an existing reader_ptr/writer_ptr is
* not completed
* 2. check if we can force the termination of a X_TO_N request when
* some data transfer took place
* 3. check if we have to cancel a timer when the (first) data has been
* Xferred
* 4. Check if we have to kick out a queued request because its
* processing is really blocked (for some reason)
*/
if (reader_ptr && writer_ptr) {
__ASSERT_NO_MSG(!(TERM_XXX & reader_ptr->args.pipe_xfer_req.status) &&
!(TERM_XXX & writer_ptr->args.pipe_xfer_req.status));
/*
* this could be possible when data Xfer operations are jammed
* (out of data Xfer resources e.g.)
*/
/* later on, at least one of them will be completed.
* Force termination of X_TO_N request?
* - If one of the requesters is X_TO_N and the other one is
* ALL_N, we cannot force termination of the X_TO_N request
* - If they are both X_TO_N, we can do so (but can this
* situation be?)
* In this version, we will NOT do so and try to transfer data
* as much as possible as there are now 2 parties present to
* exchange data, possibly directly (this is an implementation
* choice, but I think it is best for overall application
* performance)
*/
;
} else if (reader_ptr) {
__ASSERT_NO_MSG(!(TERM_XXX & reader_ptr->args.pipe_xfer_req.status));
/* check if this lonely reader is really blocked, then we will
* delist him (if he was listed uberhaupt) == EMERGENCY BREAK
*/
if (ReaderInProgressIsBlocked(pipe_ptr, reader_ptr)) {
if (_X_TO_N & _k_pipe_option_get(&reader_ptr->args) &&
(reader_ptr->args.pipe_xfer_req.xferred_size != 0)) {
_k_pipe_request_status_set(&reader_ptr->args.pipe_xfer_req,
TERM_SATISFIED);
} else {
/* in all other cases: forced termination */
_k_pipe_request_status_set(&reader_ptr->args.pipe_xfer_req,
TERM_FORCED);
}
if (reader_ptr->head) {
DeListWaiter(reader_ptr);
myfreetimer(&(reader_ptr->Time.timer));
}
if (reader_ptr->args.pipe_xfer_req.num_pending_xfers == 0) {
reader_ptr->Comm = _K_SVC_PIPE_GET_REPLY;
/*
* if terminated and no pending Xfers anymore,
* we have to reply
*/
_k_pipe_get_reply(reader_ptr);
}
} else {
/*
* temporary stall (must be, processing will continue
* later on)
*/
}
} else if (writer_ptr) {
__ASSERT_NO_MSG(!(TERM_SATISFIED & writer_ptr->args.pipe_xfer_req.status));
/*
* check if this lonely Writer is really blocked, then we will
* delist him (if he was listed uberhaupt) == EMERGENCY BREAK
*/
if (WriterInProgressIsBlocked(pipe_ptr, writer_ptr)) {
if (_X_TO_N & _k_pipe_option_get(&writer_ptr->args) &&
(writer_ptr->args.pipe_xfer_req.xferred_size != 0)) {
_k_pipe_request_status_set(&writer_ptr->args.pipe_xfer_req,
TERM_SATISFIED);
} else {
/* in all other cases: forced termination */
_k_pipe_request_status_set(&writer_ptr->args.pipe_xfer_req,
TERM_FORCED);
}
if (writer_ptr->head) {
DeListWaiter(writer_ptr);
myfreetimer(&(writer_ptr->Time.timer));
}
if (writer_ptr->args.pipe_xfer_req.num_pending_xfers == 0) {
writer_ptr->Comm = _K_SVC_PIPE_PUT_REPLY;
/*
* if terminated and no pending Xfers anymore,
* we have to reply
*/
_k_pipe_put_reply(writer_ptr);
}
} else {
/*
* temporary stall (must be, processing will continue
* later on)
*/
}
} else {
__ASSERT_NO_MSG(1 == 0); /* we should not come ... here :-) */
}
/* check if we have to cancel a timer for a request */
#ifdef CANCEL_TIMERS
if (reader_ptr) {
if (reader_ptr->args.pipe_xfer_req.xferred_size != 0) {
if (reader_ptr->head) {
myfreetimer(&(reader_ptr->Time.timer));
/* do not delist however */
}
}
}
if (writer_ptr) {
if (writer_ptr->args.pipe_xfer_req.xferred_size != 0) {
if (writer_ptr->head) {
myfreetimer(&(writer_ptr->Time.timer));
/* do not delist however */
}
}
}
#endif
}