subsystem.c

This a framework to test new ideas in transmission technology. Actual development is a LDPC-coder in

/***************************************************************************
             subsystem.c  -  RTLinux kernel module for subsystem
                            -------------------
    begin                :  2002
    authors              :  Linus Gasser
    emails               :  [e-mail]
 ***************************************************************************/

/***************************************************************************
                                 Changes
                                 -------
 date - name - description
 02/09/12 - ineiti - begin
 02/10/22 - ineiti - passing of whole data-block w/o freeing
 02/12/19 - ineiti - fix allocation of data-blocks
 03/01/06 - ineiti - also propagate changed data-blocks in ports
 03/02/28 - ineiti - cleaned up a lot w/regard to resizing, reconfig
 03/04/07 - ineiti - added profiling
 03/08/18 - ineiti - added threading-support
 03/08/19 - ineiti - added tracking-support
 03/11/13 - ineiti - added SUBS_MSG_UPDATE which reconfigures all necessary
                     modules
 03/12/15 - ineiti - ports are also checked for changes upon receiving
                     user-messages
 04/03/22 - ineiti - fixed tracking-stuff
 04/03/22 - ineiti - renamed SUBS_MSG_UPDATE to SUBS_MSG_PREPARE
 04/03/31 - ineiti - when sending a swr_sdb_send_msg, one can now give a
                     return parameter
 04/03/31 - ineiti - MSG_PREPARE returns SUBS_STATS_WORKING if one of the
                     module in the chain is still working
 04/04/20 - ineiti - cleaning up RECONFIG-calls 
 04/05/12 - ineiti - only profile on successful data-proceeding
 04/06/01 - ineiti - added SUBS_STATUS_MULTI_IN to define a module that
                     needs all inputs before it can process

 **************************************************************************/

/***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 ***************************************************************************/


/**
 * This is the implementation of the actual thread of the subsystem.
 * In this place the message-queue is polled and appropriate action
 * is taken here.
 */


/**
 * Debugging priorities
 *
 * LVL 0 : Error messages only
 * LVL 1 : Important status messages
 * LVL 2 : Not too much traffic
 * LVL 3 : A bit more detailed
 * LVL 4 : S P A M - all we can do
 *
 * This way you can increase or decrease the level of debugging
 * messages depending on what you're doing.
 */

#define DBG_LVL 0

#include "system.h"
#include "debugging.h"
#include "memory.h"
#include "sdb.h"
#include "subsystem.h"
#include "connection_local.h"
#include "dbg.h"
#include "parameter_types.h"


void *malloc_port( swr_port_t *port ) {
  int size = port->size * swr_spc_get_type_size( port->type );
  port->data = swr_malloc( size );
  PR_DBG( 4, "Malloc for %i bytes at address: %p\n", size, port->data );
  return ( port->data );
}


/**
 * Calls the reconfiguration-function and reconfigures the in/outputs, too...
 */
void reconfig( swr_sdb_t *context ) {
  int i, conf = 0;
  func( context->cdb_desc->fn_reconfigure );

  if ( IS_STATUS( RESIZE_UP ) ) {
    // If at least one output-port is non-zero, we go for reconfig
    if ( ( i = context->cdb_desc->outputs.nbr_ports ) > 0 ) {
      for ( conf=0 ; i > 0; i-- ) {
        conf |= context->port_out[i-1].size;
      }
    }
    if ( conf ) {
      PR_DBG( 4, "Reconfigure inputs due to reconfig\n" );
      func( context->cdb_desc->fn_configure_inputs );
    }
  }

  if ( IS_STATUS( RESIZE_DOWN ) ) {
    // If at least one input-port is non-zero, we go for reconfig
    if ( ( i = context->cdb_desc->inputs.nbr_ports ) > 0 ) {
      for ( conf=0 ; i > 0; i-- ) {
        conf |= context->port_in[i-1].size;
      }
    }
    if ( conf ) {
      PR_DBG( 4, "Reconfigure outputs due to reconfig\n" );
      func( context->cdb_desc->fn_configure_outputs );
    }
  }

  CLR_STATUS( RECONF );
}


int prepare( swr_sdb_t *context, long int *t_prep ){
  int port, ret = 0;
  
  if ( IS_STATUS( PREPARE_SWALLOW ) ){
    PR_DBG( 3, "Swallowing prepare-msg in module %i\n", context->id );
    return 0;
  }

  if ( IS_STATUS( WORKING ) ){
    PR_DBG( 3, "Preparing while working in module %i\n",
	    context->id );
    return STATUS( WORKING );
  }
  
  if ( IS_STATUS( PREPARE ) ){
    PR_DBG( 2, "Loop in the chain!\n" );
    return 0;
  }
  SET_STATUS( PREPARE );

  // Check for time
  if ( !t_prep ){
    // If it's the first call, get the time
    context->time_prepare = get_time_usec();
    PR_DBG( 4, "Time_prepare: %li\n", context->time_prepare );
    t_prep = &context->time_prepare;
  } else {
    // update the time
    context->time_prepare = *t_prep;
  }

  // Call eventually the reconfiguration-part
  if ( IS_STATUS( RECONF ) ) {
    PR_DBG( 3, "Reconfiguring during prepare\n" );
    // By sending a message, we make sur that an eventual size-change to
    // an input/output port is taken into account
    swr_sdb_send_msg( context->id, SUBS_MSG_RECONFIG, NULL, -1 );
  }

  // Propagate the prepare-message to the rest of the chain
  for ( port=0; port<context->cdb_desc->outputs.nbr_ports; port++ ){
    if ( context->port_out[port].sdb_id >= 0 ){
      PR_DBG( 4, "Sending prepare through port %i to %i\n", port,
	      context->port_out[port].sdb_id );
      ret = swr_sdb_send_msg( context->port_out[port].sdb_id, 
			      SUBS_MSG_PREPARE, t_prep, -1 );
    }
  }

  CLR_STATUS( PREPARE );
  return ret;
}


/**
 * Test whether we have to malloc something
 */
void evtl_malloc( swr_sdb_t *context,
                  void *data, swr_port_t *port_rm, int port_nbr ) {

  // free eventually the data-part
  if ( port_rm->data &&
       ( port_rm->flags & SWR_PORT_THIS_FREE ) ) {
    PR_DBG( 4, "Freeing port %i\n", port_nbr );
    if ( swr_free( port_rm->data ) > 0 ) {
      PR_DBG( 0, "Error while freeing port %i from id %i\n",
              port_nbr, context->id );
    }
    port_rm->flags &= ~SWR_PORT_THIS_FREE;
  } else if ( ( port_rm->flags & SWR_PORT_OTHER_FREE ) &&
              !( port_rm->flags & SWR_PORT_PASSED_THROUGH ) ) {
    port_rm->data = NULL;
    port_rm->flags &= ~SWR_PORT_OTHER_FREE;
  }

  // And allocate some space
  if ( data ) {
    // Take the space given by the receiver
    if ( port_rm->flags & SWR_PORT_OWN_MALLOC ) {
      PR_DBG( 0, "Overwriting own malloc in port %i id %i!\n",
              port_nbr, context->id );
    }
    PR_DBG( 4, "Copying propagation data-pointer: %p\n", data );
    port_rm->data = data;
    port_rm->flags |= SWR_PORT_OTHER_FREE;
    port_rm->flags &= ~SWR_PORT_THIS_FREE;
  } else if ( !( port_rm->flags & SWR_PORT_OWN_MALLOC ) ) {
    if ( port_rm->flags & SWR_PORT_PASSED_THROUGH ) {
      PR_DBG( 4, "Passed the data-pointer\n" );
      port_rm->flags |= SWR_PORT_OTHER_FREE;
    } else if ( ( port_rm->sdb_id >= 0 ) &&
                !( port_rm->flags & SWR_PORT_OTHER_MALLOC ) ) {
      // Allocate an own space
      PR_DBG( 4, "Allocating own space\n" );
      malloc_port( port_rm );
      port_rm->flags |= SWR_PORT_THIS_FREE;
    } else {
      PR_DBG( 4, "Not allocating yet\n" );
    }
  }
}

/**
 * Sends a resize-message
 */
void send_resize( swr_sdb_id to, swr_msgq from, swr_port_t *port,
                  swr_signal_dir_t dir ) {
  swr_propagation_t *prop;

  prop = (swr_propagation_t*)swr_malloc(sizeof(swr_propagation_t));
  prop->size = port->size;
  prop->port = port->port;
  prop->data = port->data;
  prop->flags = port->flags;
  prop->dir_to = dir;
  PR_DBG( 4, "Sending resize: port %i to size %i, "
          "data %p, sdb_id:%i, flags: 0x%x\n",
          prop->port, prop->size, prop->data, to, prop->flags );
  swr_sdb_send_msg( to, SUBS_MSG_RESIZE, prop, from );
}


/**
 * Connects either an input or an output port to another module
 */
void conn_msg( struct swr_sdb_t_s *context, swr_conn_req_t *conn ) {
  struct swr_port_t_s *this_port = NULL;

  if ( conn->dir == OUTPUT ) {
    // To connect with this_port output-port
    PR_DBG( 3, "connect: <%i:%i> -> %i:%i\n", context->id, conn->port_out,
            conn->sdb_id, conn->port_in );
    this_port = &context->port_out[ conn->port_out ];
    this_port->port = conn->port_in;
  }

  if (conn->dir == INPUT) {
    // To connect with this_port input-port
    PR_DBG( 3, "connect: %i:%i -> <%i:%i>\n", conn->sdb_id, conn->port_out,
            context->id, conn->port_in );
    this_port = &context->port_in[ conn->port_in ];
    this_port->port = conn->port_out;
  }

  if (this_port == NULL) {
    PR_DBG(0, "conn_msg: error\n");
    return;
  }
  if ( conn->other_flags & SWR_PORT_OWN_MALLOC ) {
    this_port->flags |= SWR_PORT_OTHER_MALLOC;
  }

  this_port->sdb_id = conn->sdb_id;

  // send a propagation message if the size of "this_port" is > 0
  if (this_port->size > 0) {
    conn->prop = (swr_propagation_t*)swr_malloc(sizeof(swr_propagation_t));
    conn->prop->size = this_port->size;
    conn->prop->port = this_port->port;
    conn->prop->dest_sdb = conn->sdb_id;
    conn->prop->dir_to = conn->dir == OUTPUT ? INPUT : OUTPUT;
    PR_DBG( 3, "This flags: 0x%x - Other flags: 0x%x\n", this_port->flags,
            conn->other_flags );
    if ( !( this_port->flags & SWR_PORT_OWN_MALLOC ) &&
         !( this_port->data ) &&
         !( conn->other_flags & SWR_PORT_OWN_MALLOC ) ) {
      // It's this function's job to allocate the port
      PR_DBG( 4, "Allocating in conn_msg\n" );
      malloc_port( this_port );
      this_port->flags |= SWR_PORT_THIS_FREE;
    }
    conn->prop->data = this_port->data;
    conn->prop->flags = this_port->flags;
    PR_DBG( 3, "Sending conn-resize: port %i to size %i, sdb:%i\n",
            conn->prop->port, conn->prop->size, conn->prop->dest_sdb );
    // The message will be retrieved by conn_send in connection_local.h
  }
}



/**
 * Destroys a connection
 */
void disconn_msg( swr_sdb_t *context, swr_conn_req_t *conn ) {
  struct swr_port_t_s *this_port = NULL;

  if ( conn->dir == OUTPUT ) {
    this_port = &context->port_out[ conn->port_out ];
    PR_DBG( 3, "disconnect: <%i:%i> -> %i:%i\n", context->id, conn->port_out,
            conn->sdb_id, conn->port_in );
  } else {
    this_port = &context->port_in[ conn->port_in ];
    PR_DBG( 3, "disconnect: %i:%i -> <%i:%i>\n", conn->sdb_id, conn->port_out,
            context->id, conn->port_in );
  }
  this_port->sdb_id = -1;
  // And free the data
  if ( this_port->flags & SWR_PORT_THIS_FREE ) {
    PR_DBG( 3, "Freeing data-port\n" );
    if ( swr_free( this_port->data ) ) {
      PR_DBG( 0, "Error in freeing port (in/out) %i/%i with dir %s and id %i\n",
              conn->port_in, conn->port_out,
              conn->dir ? "out(1)" : "in(0)",
              context->id );
    }
    this_port->data = NULL;
  }
  // Delete now unused flags
  this_port->flags &= ~( SWR_PORT_OTHER_FREE + SWR_PORT_THIS_FREE +
                         SWR_PORT_OTHER_MALLOC );
}


/**
 * Executes the processing-function of the sub-module.
 */
int rcv_data( swr_sdb_t *context, int ret, void *data ) {
  int r = 0;

  func_ret( context->cdb_desc->fn_process_data, r );
  return r;
}


/**
 * Looks if data is available at an output and sends it, also restoring
 * the ports to NULL for further data
 */
void send_data( swr_sdb_t *context ) {
  int port;