demo_ha_blink.c

// "High availability" blinker demo.
//
// Uses a remote blinker if accessible, local blinker otherwise.
//
// "Producer" device:
//
// `ha_blinker` sends a message (desired number of blinks) to one of the
// channels from the `channels` array. The first channel is a remote (CAN)
// blinker, the second is a local blinker => if the remote blinker fails, local
// blinker is used as a failover.
//
// `can_blinker` reads from a channel of numbers and writes to a channel of CAN
// messages (CAN server tx channel typically).
//
// "Consumer" device:
//
// `consumer` reads CAN messages from a channel (typically a CAN server rx
// channel) and writes to a channel of numbers (typically a local "blinker"
// server channel).
//
// BUG: You will see some interweaved blinks on both "producer" and "consumer"
//      device after the "consumer" device goes back online. This is due to a
//      CAN buffer on STM32 and enabled auto-resend of the CAN messages.

#include <o1heap.h>

#include "csp.h"
#include "hal.h"

#define CAN_MSG_ID 123

extern O1HeapInstance *heap;

typedef struct {
  chan_t **channels;
  uint8_t channels_len;
  uint8_t blinks_num;
  tick_t delay;
} ha_blinker_state_t;

typedef struct {
  chan_t *input_c;
  chan_t *can_tx_c;
} can_blinker_state_t;

typedef struct {
  chan_t *can_rx_c;
  chan_t *blinker_c;
} consumer_state_t;

static void ha_blinker_send(task_data_t task_data);
static void ha_blinker_delay(task_data_t task_data);
static void can_blinker_send(chan_data_t data, task_data_t task_data);
static void can_blinker_recv(task_data_t task_data);
static void consumer_send(chan_data_t data, task_data_t task_data);
static void consumer_recv(task_data_t task_data);

// -------------------------------------------------------- API ----------------

// To be started on the "publisher" device.
// can_tx_c = CAN server tx channel
// blinker_c = local blinker channel
// blinks_num = desired number of blinks
// delay = desired delay between blinks (in ticks)
void start_test_ha_blinker_publisher(chan_t *can_tx_c, chan_t *blinker_c,
                                     uint8_t blinks_num, tick_t delay) {
  static chan_t can_blinker_c;
  static chan_t *channels[2];

  // remote blinker channel
  channels[0] = &can_blinker_c;
  // local blinker channel
  channels[1] = blinker_c;

  static ha_blinker_state_t ha_blinker_state;
  ha_blinker_state.channels = channels;
  ha_blinker_state.channels_len = 2;
  ha_blinker_state.blinks_num = blinks_num;
  ha_blinker_state.delay = delay;
  csp_spawn(ha_blinker_send, &ha_blinker_state);

  csp_chan_init(&can_blinker_c);
  static can_blinker_state_t can_blinker_state;
  can_blinker_state.can_tx_c = can_tx_c;
  can_blinker_state.input_c = &can_blinker_c;
  csp_recv(&can_blinker_c, can_blinker_send, &can_blinker_state);
}

// To be started on the "consumer" device.
// can_rx_c = CAN server rx channel
// blinker_c = local blinker channel
void start_test_ha_blinker_consumer(chan_t *can_rx_c, chan_t *blinker_c) {
  static consumer_state_t consumer_state;
  consumer_state.can_rx_c = can_rx_c;
  consumer_state.blinker_c = blinker_c;
  csp_recv(can_rx_c, consumer_send, &consumer_state);
}

// -------------------------------------------------------- implementation -----

static void ha_blinker_send(task_data_t task_data) {
  ha_blinker_state_t *state = (ha_blinker_state_t *)task_data;

  // Group is like "select" in Go
  // => message will be sent to the first ready channel only. All other
  // send tasks will be deleted then.
  group_id_t grp = csp_get_free_group_id();
  for (uint8_t i = 0; i < state->channels_len; i++) {
    csp_send_grp(state->channels[i], (chan_data_t)state->blinks_num,
                 ha_blinker_delay, task_data, grp);
  }
}

static void ha_blinker_delay(task_data_t task_data) {
  ha_blinker_state_t *state = (ha_blinker_state_t *)task_data;
  csp_delay(state->delay, ha_blinker_send, task_data);
}

static void can_blinker_send(chan_data_t data, task_data_t task_data) {
  can_blinker_state_t *state = (can_blinker_state_t *)task_data;

  can_msg_t *msg;
  if ((msg = o1heapAllocate(heap, sizeof(can_msg_t))) == NULL) {
    hal_panic(PANIC_OUT_OF_MEMORY);
  }
  msg->id = CAN_MSG_ID;
  msg->payload[0] = (uint8_t)data;
  msg->len = 1;

  csp_send(state->can_tx_c, (chan_data_t)msg, can_blinker_recv, task_data);
}

static void can_blinker_recv(task_data_t task_data) {
  can_blinker_state_t *state = (can_blinker_state_t *)task_data;
  csp_recv(state->input_c, can_blinker_send, task_data);
}

static void consumer_send(chan_data_t data, task_data_t task_data) {
  consumer_state_t *state = (consumer_state_t *)task_data;

  can_msg_t *msg = (can_msg_t *)data;

  if (msg->id != CAN_MSG_ID || msg->len != 1) {
    // Ignore invalid/unexpected CAN message. Go straight to RECV again.
    o1heapFree(heap, msg);
    consumer_recv(task_data);
    return;
  }

  dbg_print_s("[CAN RX] ");
  dbg_print_unl(msg->payload[0]);

  csp_send(state->blinker_c, (chan_data_t)msg->payload[0], consumer_recv,
           task_data);
  o1heapFree(heap, msg);
}

static void consumer_recv(task_data_t task_data) {
  consumer_state_t *state = (consumer_state_t *)task_data;
  csp_recv(state->can_rx_c, consumer_send, task_data);
}