--- /dev/null
+/*
+ * FastCommunication.cpp / Version PPE
+ *
+ * Created on: Mar 3, 2009
+ * Author: lschor
+ *
+ * For remarks to this file, please refer to the SPE variant.
+ *
+ * Completion of the SPEs:
+ * Is coordinated by this process. Each SPE has to send a request to
+ * the PPE if it is allowed to finish.
+ * If all SPEs have requested to complete, the PPE sends to the SPEs a
+ * message that they can complete, thus return.
+ */
+
+#include "FastCommunication.h"
+
+/*
+ * Constructor
+ */
+FastCommunication::FastCommunication(int nrOfQueues,
+ uint64_t *context_all, uint64_t * ea_ls_base,
+ int32_t * queueFromSPEIn, int32_t * queueOnSPEIn,
+ uint64_t *fifoTails) {
+
+ _context_all = context_all;
+ queueFromSPE = queueFromSPEIn;
+ queueOnSPE = queueOnSPEIn;
+ _ea_ls_base = ea_ls_base;
+ _fifoTails = fifoTails;
+ _nrSpeComplete = 0;
+
+ _nrOfQueues = nrOfQueues;
+ try { _fifos = new fifoCollection[nrOfQueues]; }
+ catch(std::bad_alloc &e) {
+ fprintf(stderr, "[FastCommunication] Memory allocation failure\n");
+ exit(1);
+ }
+
+ for (int i = 0; i < MAXNOREQ; i++) {
+ _request[i].valid = false;
+ }
+}
+
+/*
+ * Deconstructor
+ */
+FastCommunication::~FastCommunication() {
+ delete _fifos;
+}
+
+/*
+ * Register an additional FIFO in the Communication
+ */
+bool FastCommunication::addFifo(int fifoNr, Fifo* fifo, int type,
+ int queue) {
+ _fifos[fifoNr].fifo = fifo;
+ _fifos[fifoNr].queue = queue;
+ _fifos[fifoNr].type = type;
+ _fifos[fifoNr].iswfifo = false;
+
+ // Per FIFO queue, we need two requests, but we can store at max MAXNOREQ of them
+ if (_nrOfRequest < MAXNOREQ) {
+ _nrOfRequest += 2;
+ }
+
+ return true;
+}
+
+/*
+ * Register an additional WindowedFIFO in the Communication
+ */
+bool FastCommunication::addWFifo(int fifoNr, WindowedFifo* wfifo,
+ int type, int queue) {
+ _fifos[fifoNr].wfifo = wfifo;
+ _fifos[fifoNr].queue = queue;
+ _fifos[fifoNr].type = type;
+ _fifos[fifoNr].iswfifo = true;
+
+ // Per FIFO queue, we need two requests, but we can store at max MAXNOREQ of them
+ if (_nrOfRequest < MAXNOREQ) {
+ _nrOfRequest += 2;
+ }
+
+ return true;
+}
+
+/*
+ * Communication: Main update procedure to update
+ */
+bool FastCommunication::update() {
+ // Check if you have received a new message
+ for (int processNr = 0; processNr < NUM_SPES; processNr++) {
+ if (!spe_out_mbox_status((spe_context*) _context_all[processNr]))
+ continue;
+
+ uint32_t messageIn;
+ spe_out_mbox_read((spe_context*) _context_all[processNr],
+ &messageIn, 1);
+
+ uint32_t code = GETFASTCODE(messageIn);
+ uint32_t queue = GETFASTQUEUE(messageIn);
+ uint32_t len = GETFASTLEN(messageIn);
+
+ /***************************************************************************************************************/
+ if (code == SPE_READ_DEMAND) // Have to send a write address (1) --> (2)
+ {
+ // Find out for which fifo the request is
+ fifoCollection *fifocol = NULL;
+ int i;
+ for (i = 0; i < _nrOfQueues; i++) {
+ if (_fifos[i].queue == queue) {
+ fifocol = &_fifos[i];
+ break;
+ }
+ }
+
+ // The queue was not found
+ if (fifocol == NULL) {
+ printf("PPU COM> ERROR, this queue does not exists!\n");
+ return false;
+ }
+
+ // %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+ // Windowed FIFO
+ if (_fifos[i].iswfifo) {
+ WindowedFifo* wfifo = NULL;
+ wfifo = _fifos[i].wfifo;
+
+ uint32_t inTail = wfifo->_inTail;
+
+ // This is the len we like to read
+ len = len > (wfifo->unused()) ? wfifo->unused() : len;
+
+ // We can read something
+ if (len > 0) {
+ // Write all information we used to the request-memory
+ comRequest* request = newRequest();
+
+ // Has no memory to store a new request
+ if (request == NULL) {
+ // not possible to start a request for this SPE --> Send len = 0
+ uint32_t message = CREATEFASTMESSAGE(
+ SPE_READ_COMPLETE, queue, 0);
+ sendMessage(message, queueFromSPE[queue]);
+ return false;
+ }
+
+ uint64_t baseAddress = _fifoTails[queue] + inTail;
+ while (baseAddress % ALIGNMENT_FACTOR_POWER2 != 0)
+ baseAddress--;
+
+ // Offset
+ uint32_t offset = _fifoTails[queue] + inTail
+ - baseAddress;
+
+ // Store all data in the request buffer
+ request->data = (char *) _malloc_align(roundDMA(len
+ + offset), ALIGNMENT_FACTOR);
+ request->len = len;
+ request->wfifo = wfifo;
+ request->iswfifo = true;
+ request->status = read_started;
+ request->queue = queue;
+ request->offset = offset;
+
+ int ret;
+ do {
+ ret = spe_mfcio_put(
+ (spe_context*) _context_all[processNr],
+ baseAddress, (void *) &(request->data[0]),
+ roundDMA(len + offset), request->tag_id,
+ 0, 0);
+ } while (ret != 0);
+ } else {
+#ifndef STORE_REQUESTS // Send len = 0 back to the sender, this one should try it in a later phase
+ uint32_t message = CREATEFASTMESSAGE(
+ SPE_READ_COMPLETE, queue, 0);
+ sendMessage(message, queueFromSPE[queue]);
+#else // I store the request and may try in a later time to start the transfer
+ comRequest* request = newRequest();
+
+ // Has no memory to store a new request
+ if (request == NULL)
+ {
+ // not possible to start a request for this SPE --> Send len = 0
+ uint32_t message = CREATEFASTMESSAGE(SPE_READ_COMPLETE, queue, 0);
+ sendMessage(message, queueFromSPE[queue]);
+ return false;
+ }
+
+ request->len = len;
+ request->wfifo = wfifo;
+ request->iswfifo = true;
+ request->status = read_pending;
+ request->queue = queue;
+#endif
+ }
+
+ }
+
+ // %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+ // Classic FIFO
+ else {
+ Fifo* fifo = NULL;
+ fifo = _fifos[i].fifo;
+
+ uint32_t inTail = fifo->_inTail;
+
+ // This is the len we like to read
+ len = len > (fifo->unused()) ? fifo->unused() : len;
+
+ // We can read something
+ if (len > 0) {
+ // Write all information we used to the request-memory
+ comRequest* request = newRequest();
+
+ // Has no memory to store a new request
+ if (request == NULL) {
+ // not possible to start a request for this SPE --> Send len = 0
+ uint32_t message = CREATEFASTMESSAGE(
+ SPE_READ_COMPLETE, queue, 0);
+ sendMessage(message, queueFromSPE[queue]);
+ return false;
+ }
+
+ uint64_t baseAddress = _fifoTails[queue] + inTail;
+ while (baseAddress % ALIGNMENT_FACTOR_POWER2 != 0)
+ baseAddress--;
+
+ // Offset
+ uint32_t offset = _fifoTails[queue] + inTail
+ - baseAddress;
+
+ // Store all data in the request buffer
+ request->data = (char *) _malloc_align(roundDMA(len
+ + offset), ALIGNMENT_FACTOR);
+ request->len = len;
+ request->fifo = fifo;
+ request->iswfifo = false;
+ request->status = read_started;
+ request->queue = queue;
+ request->offset = offset;
+
+ int ret;
+ do {
+ ret = spe_mfcio_put(
+ (spe_context*) _context_all[processNr],
+ baseAddress, (void *) &(request->data[0]),
+ roundDMA(len + offset), request->tag_id,
+ 0, 0);
+ } while (ret != 0);
+ }
+
+ // Len == 0
+ else {
+#ifndef STORE_REQUESTS // Send len = 0 back to the sender, this one should try it in a later phase
+ uint32_t message = CREATEFASTMESSAGE(
+ SPE_READ_COMPLETE, queue, 0);
+ sendMessage(message, queueFromSPE[queue]);
+#else // I store the request and may try in a later time to start the transfer
+ comRequest* request = newRequest();
+
+ // Has no memory to store a new request
+ if (request == NULL) {
+ // not possible to start a request for this SPE --> Send len = 0
+ uint32_t message = CREATEFASTMESSAGE(SPE_READ_COMPLETE, queue, 0);
+ sendMessage(message, queueFromSPE[queue]);
+ return false;
+ }
+
+ request->len = len;
+ request->fifo = fifo;
+ request->iswfifo = false;
+ request->status = read_pending;
+ request->queue = queue;
+#endif
+ }
+ }
+ }
+
+ /***************************************************************************************************************/
+ else if (code == SPE_READ_COMPLETE) // Can finish a write request (4) --> (5)
+ {
+ // Get the stored request
+ comRequest* request = getRequest(read_request_sent, queue);
+
+ if (request == NULL) {
+ printf("PPU Communicate> Couldn't find the request\n");
+ return 0;
+ }
+
+ // Inform my FIFO that the request is completed
+ if (request->iswfifo)
+ request->wfifo->dmaRead(len);
+ else
+ request->fifo->dmaRead(len);
+
+ // Request free
+ deleteRequest(request);
+ }
+
+ /***************************************************************************************************************/
+ else if (code == SPE_COMPLETE) // One SPE has finished
+ {
+ _nrSpeComplete++;
+ }
+ }
+
+ /***************************************************************************************************************/
+ // Check if some active write processes have finished (4)
+ uint8_t req = _currentRequest;
+
+ for (int i = 0; i < _nrOfRequest; i++) {
+ if (_request[req].valid) {
+ if (_request[req].status == read_started) {
+ if (!(testMessage(queueFromSPE[_request[req].queue]) > 0))
+ continue;
+
+ uint32_t tag_id = _request[req].tag_id;
+ uint32_t ret;
+ uint32_t
+ test =
+ spe_mfcio_tag_status_read(
+ (spe_context*) _context_all[queueFromSPE[_request[req].queue]],
+ 0, SPE_TAG_IMMEDIATE, &ret);
+
+ if (((ret & (1 << tag_id)) == 0)) {
+ // Have to write the data into the fifo
+ if (_request[req].iswfifo) { // WFIFO
+
+ _request[req].wfifo->dmaWrite(
+ (char *) _request[req].data
+ + _request[req].offset,
+ _request[req].len);
+ _free_align(_request[req].data);
+
+ // Increase the inTail value (used to know where the last request was started)
+
+ _request[req].wfifo->_inTail
+ = (_request[req].wfifo->_inTail
+ + _request[req].len) % (FIFO_SIZE[_request[req].queue]);
+ } else { // FIFO
+ _request[req].fifo->write(
+ (char *) _request[req].data
+ + _request[req].offset,
+ _request[req].len);
+ _free_align(_request[req].data);
+
+ _request[req].fifo->_inTail
+ = (_request[req].fifo->_inTail
+ + _request[req].len) % (FIFO_SIZE[_request[req].queue]);
+
+ }
+
+ uint32_t message = CREATEFASTMESSAGE(
+ SPE_READ_COMPLETE, _request[req].queue,
+ _request[req].len);
+ sendMessage(message, queueFromSPE[_request[req].queue]);
+
+ deleteRequest(&_request[req]);
+ _currentRequest = (req + 1) % _nrOfRequest;
+
+ break;
+ }
+ } else if (_request[req].status == read_pending) { // Has an open request for sending
+
+ // %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+ // Windowed FIFO
+ if (_request[req].iswfifo) {
+
+ if (_request[req].wfifo->unused() > 0) {
+ comRequest *request = &_request[req];
+
+ // This is the len we like to read
+ uint32_t len = request->len > (request->wfifo->unused()) ? request->wfifo->unused()
+ : request->len;
+
+ uint32_t inTail = request->wfifo->_inTail;
+ uint64_t baseAddress = _fifoTails[request->queue]
+ + inTail;
+
+ while (baseAddress % ALIGNMENT_FACTOR_POWER2 != 0)
+ baseAddress--;
+
+ // Offset --> How much we had to align
+ uint32_t offset = _fifoTails[request->queue]
+ + inTail - baseAddress;
+
+ request->data = (char *) _malloc_align(roundDMA(
+ len + offset), ALIGNMENT_FACTOR);
+ request->len = len;
+ request->status = read_started;
+ request->offset = offset;
+
+ // Set up the request in the MFC
+ int ret;
+ do {
+ ret = spe_mfcio_put(
+ (spe_context*) _context_all[queueFromSPE[_request[req].queue]],
+ baseAddress,
+ (void *) &(request->data[0]),
+ roundDMA(len + offset),
+ request->tag_id, 0, 0);
+ } while (ret != 0);
+ }
+ }
+
+ // %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+ // Classic FIFO
+ else {
+
+ if (_request[req].fifo->unused() > 0) {
+ comRequest *request = &_request[req];
+
+ // This is the len we like to read
+ uint32_t len = request->len > (request->fifo->unused()) ? request->fifo->unused() : request->len;
+
+ uint32_t inTail = request->fifo->_inTail;
+ uint64_t baseAddress = _fifoTails[request->queue]
+ + inTail;
+
+ while (baseAddress % 16 != 0)
+ baseAddress--;
+
+ // Offset
+ uint32_t offset = _fifoTails[request->queue]
+ + inTail - baseAddress;
+
+ // Store all data in the request buffer
+ request->data = (char *) _malloc_align(roundDMA(
+ len + offset), ALIGNMENT_FACTOR);
+ request->len = len;
+ request->status = read_started;
+ request->offset = offset;
+
+ // Start the DMA transfer
+ int ret;
+ do {
+ ret
+ = spe_mfcio_put(
+ (spe_context*) _context_all[queueFromSPE[_request[req].queue]],
+ baseAddress,
+ (void *) &(request->data[0]),
+ roundDMA(len + offset),
+ request->tag_id, 0, 0);
+ } while (ret != 0);
+ }
+ }
+ }
+ }
+ req = (req + 1) % _nrOfRequest;
+ }
+
+ /***************************************************************************************************************/
+ // Have some out - queue some data to send
+ for (int i = 0; i < _nrOfQueues; i++) {
+ if (_fifos[i].type == this->out) {
+ if (_fifos[i].iswfifo) { // WFIFO
+ // Start only a write process if there is really enough place in the outbound mailbox
+ if (_fifos[i].wfifo->dmaAllowed()
+ && _fifos[i].wfifo->used() > 0) {
+ uint32_t queue = _fifos[i].queue;
+
+ // Can we send a message to this processor or is it blocked?
+ if (testMessage(queueOnSPE[queue]) <= 0) {
+ continue;
+ }
+
+ // Write all information we used to the request-memory
+ comRequest* request = newRequest();
+
+ // Has no memory to store a new request
+ if (request == NULL) {
+ } else {
+ uint32_t len = _fifos[i].wfifo->dmaStart();
+ // Create a write-demand message
+ uint32_t message = CREATEFASTMESSAGE(
+ SPE_READ_DEMAND, queue, len);
+
+ request->len = len;
+ request->queue = queue;
+ request->status = read_request_sent;
+ request->wfifo = _fifos[i].wfifo;
+ request->iswfifo = true;
+
+ sendMessage(message, queueOnSPE[queue]);
+ }
+ break;
+ }
+ } else { // FIFO
+ // Start only a write process if there is really enough place in the outbound mailbox
+ if (_fifos[i].fifo->dmaAllowed() && _fifos[i].fifo->used()
+ > 0) {
+ uint32_t queue = _fifos[i].queue;
+
+ if (testMessage(queueOnSPE[queue]) <= 0)
+ continue;
+
+ // Write all information we used to the request-memory
+ comRequest* request = newRequest();
+
+ // Has no memory to store a new request
+ if (request == NULL) {
+ } else {
+ uint32_t len = _fifos[i].fifo->dmaStart();
+ uint32_t message = CREATEFASTMESSAGE(
+ SPE_READ_DEMAND, queue, len);
+ request->len = len;
+ request->queue = queue;
+ request->status = read_request_sent;
+ request->fifo = _fifos[i].fifo;
+ request->iswfifo = false;
+
+ sendMessage(message, queueOnSPE[queue]);
+ }
+ break;
+ }
+ }
+ }
+ }
+ return true;
+}
+
+/*
+ * Create a new request to store in the cache
+ *
+ */
+comRequest* FastCommunication::newRequest() {
+ for (int i = 0; i < _nrOfRequest; i++) {
+ if (!_request[i].valid) {
+ _request[i].tag_id = i + 3;
+ _request[i].valid = true;
+ return &(_request[i]);
+ }
+ }
+
+ return NULL;
+}
+
+/*
+ * Delete the request
+ *
+ */
+void FastCommunication::deleteRequest(comRequest* request) {
+ request->valid = false;
+ request->tag_id = 0;
+}
+
+/*
+ * Returns the request one like to get
+ */
+comRequest* FastCommunication::getRequest(uint8_t status, uint32_t queue) {
+ uint8_t req = _currentRequest;
+
+ for (int i = 0; i < _nrOfRequest; i++) {
+ if (_request[req].valid && _request[req].status == status
+ && _request[req].queue == queue) {
+ _currentRequest = (req + 1) % _nrOfRequest;
+ return &(_request[req]);
+ }
+ req = (req + 1) % _nrOfRequest;
+ }
+
+ return NULL;
+}
+
+/**
+ * True if no communication is necessary, false if there is active communication
+ */
+bool FastCommunication::empty() { /////////////////////////////////////////////////////////////////
+ // check if one fifo would like to send something to another SPE
+ for (int i = 0; i < _nrOfQueues; i++) {
+ if (_fifos[i].type == this->out) {
+ if (_fifos[i].iswfifo) { // WFIFO
+ if (_fifos[i].wfifo->used() > 0)
+ {
+ return false;
+ }
+ } else {
+ if (_fifos[i].fifo->used() > 0)
+ {
+ return false;
+ }
+ }
+ }
+ }
+
+ // are open sendings?
+ for (int i = 0; i < _nrOfRequest; i++) {
+ if (_request[i].valid) {
+ return false;
+ }
+ }
+
+ if (_nrSpeComplete < NUM_SPES) {
+ return false;
+ }
+
+ return true;
+}
+
+/**
+ * Send a message to a SPE
+ */
+void FastCommunication::sendMessage(uint32_t message, int32_t process) {
+ spe_in_mbox_write((spe_context*) _context_all[process],
+ (uint32_t*) &message, 1, SPE_MBOX_ANY_NONBLOCKING);
+}
+
+/**
+ * Test if we can send a message to an SPE without stalling
+ */
+int FastCommunication::testMessage(int32_t process) {
+ return spe_in_mbox_status((spe_context*) _context_all[process]);
+}
+
projects / jump.git / blobdiff