diff --git a/c_common/models/system_models/src/data_speed_up_packet_gatherer.c b/c_common/models/system_models/src/data_speed_up_packet_gatherer.c index 1ba7fc2a72..5aa4f33e66 100644 --- a/c_common/models/system_models/src/data_speed_up_packet_gatherer.c +++ b/c_common/models/system_models/src/data_speed_up_packet_gatherer.c @@ -48,35 +48,14 @@ //! max index needed to cover the chips in either direction on a spinn-5 board #define MAX_CHIP_INDEX 8 -//! SDP port commands relating to the Data In protocol -enum sdp_port_commands { - // received - //! Data In: Received message describes where to send data - SDP_SEND_DATA_TO_LOCATION_CMD = 200, - //! Data In: Received message contains data to write - SDP_SEND_SEQ_DATA_CMD = 2000, - //! Data In: Received message asks for missing sequence numbers - SDP_TELL_MISSING_BACK_TO_HOST = 2001, - // sent - //! Data In: Sent message contains missing sequence numbers - SDP_SEND_MISSING_SEQ_DATA_IN_CMD = 2002, - //! Data In: Sent message indicates that everything has been received - SDP_SEND_FINISHED_DATA_IN_CMD = 2003 -}; - //! values for port numbers this core will respond to enum functionality_to_port_num_map { - REINJECTION_PORT = 4, //!< Reinjection control messages - DATA_SPEED_UP_IN_PORT = 6 //!< Data Speed Up Inbound messages + REINJECTION_PORT = 4, //!< Reinjection control messages + DATA_COPY_IN_PORT = 7 //!< Data copy Inbound messages }; -//! threshold for SDRAM vs DTCM when allocating ::received_seq_nums_store -#define SDRAM_VS_DTCM_THRESHOLD 40000 - //! Offsets into messages enum { - //! \brief location of command IDs in SDP message - COMMAND_ID = 0, //! \brief location of where the seq num is in the packet SEQ_NUM_LOC = 0, @@ -88,69 +67,19 @@ enum { START_OF_DATA = 2 }; -//! flag when all seq numbers are missing -#define ALL_MISSING_FLAG 0xFFFFFFFE - //! mask needed by router timeout #define ROUTER_TIMEOUT_MASK 0xFF -//! Misc constants for Data In +//! Misc constants for Data Out enum { - //! offset with just command transaction id and seq in bytes - SEND_SEQ_DATA_HEADER_WORDS = 3, - //! offset with just command, transaction id - SEND_MISSING_SEQ_HEADER_WORDS = 2, - //! offset with command, transaction id, address in bytes, [x, y], max seq, - SEND_DATA_LOCATION_HEADER_WORDS = 5, //! absolute maximum size of a SDP message ABSOLUTE_MAX_SIZE_OF_SDP_IN_BYTES = 280 }; -//! Counts of items in a packet -enum { - //! \brief size of data stored in packet with command and seq - //! - //! defined from calculation - DATA_IN_NORMAL_PACKET_WORDS = - ITEMS_PER_DATA_PACKET - SEND_SEQ_DATA_HEADER_WORDS, - //! \brief size of payload for a packet describing the batch of missing - //! inbound seqs - ITEMS_PER_MISSING_PACKET = - ITEMS_PER_DATA_PACKET - SEND_MISSING_SEQ_HEADER_WORDS, -}; - //----------------------------------------------------------------------------- // TYPES AND GLOBALS //----------------------------------------------------------------------------- -//! meaning of payload in first data in SDP packet -typedef struct receive_data_to_location_msg_t { - uint command; //!< The meaning of the message - uint transaction_id; //!< The transaction that the message is taking part in - address_t address; //!< Where the stream will be writing to in memory - ushort chip_y; //!< Board-local y coordinate of chip to do write on - ushort chip_x; //!< Board-local x coordinate of chip to do write on - uint max_seq_num; //!< Maximum sequence number of data stream -} receive_data_to_location_msg_t; - -//! meaning of payload in subsequent data in SDP packets -typedef struct receive_seq_data_msg_t { - uint command; //!< The meaning of the message - uint transaction_id; //!< The transaction that the message is taking part in - uint seq_num; //!< The sequence number of this message - uint data[]; //!< The payload of real data -} receive_seq_data_msg_t; - -//! SDP packet payload definition -typedef struct sdp_msg_out_payload_t { - //! The meaning of the message - uint command; - //! The transaction associated with the message - uint transaction_id; - //! The payload data of the message - uint data[ITEMS_PER_MISSING_PACKET]; -} sdp_msg_out_payload_t; - //! the key that causes data out sequence number to be processed static uint32_t new_sequence_key = 0; @@ -172,9 +101,6 @@ static uint32_t seq_num = FIRST_SEQ_NUM; //! maximum sequence number static uint32_t max_seq_num = 0xFFFFFFFF; -//! The Data In transaction ID. Used to distinguish streams of packets -static uint32_t transaction_id = 0; - //! The Data Out transaction ID. Used to distinguish streams of packets static uint32_t data_out_transaction_id = 0; @@ -185,8 +111,8 @@ static uint32_t data[ITEMS_PER_DATA_PACKET]; //! index into ::data static uint32_t position_in_store = 0; -//! SDP message holder for transmissions -static sdp_msg_pure_data my_msg; +//! the tag to use to send SDP +static uint32_t tag = 0; //! human readable definitions of each DSG region in SDRAM enum { @@ -230,32 +156,6 @@ enum { //! LSBs (see ::key_offsets) indicate the meaning of the message. static uint data_in_mc_key_map[MAX_CHIP_INDEX][MAX_CHIP_INDEX] = {{0}}; -//! Board-relative x-coordinate of current chip being written to -static uint chip_x = 0xFFFFFFF; // Not a legal chip coordinate - -//! Board-relative y-coordinate of current chip being written to -static uint chip_y = 0xFFFFFFF; // Not a legal chip coordinate - -//! Records what sequence numbers we have received from host during Data In -static bit_field_t received_seq_nums_store = NULL; - -//! The size of the bitfield in ::received_seq_nums_store -static uint size_of_bitfield = 0; - -//! \brief Whether ::received_seq_nums_store was allocated in SDRAM. -//! -//! If false, the bitfield fitted in DTCM -static bool alloc_in_sdram = false; - -//! Count of received sequence numbers. -static uint total_received_seq_nums = 0; - -//! The most recently seen sequence number. -static uint last_seen_seq_num = 0; - -//! Where the current stream of data started in SDRAM. -static uint start_sdram_address = 0; - //! Human readable definitions of the offsets for data in multicast key //! elements. These act as commands sent to the target extra monitor core. typedef enum { @@ -310,21 +210,23 @@ static dsupg_provenance_t *sdram_prov; // FUNCTIONS //----------------------------------------------------------------------------- -//! \brief Writes the updated transaction ID to the user1 -//! \param[in] transaction_id: The transaction ID to publish -static void publish_transaction_id_to_user_1(int transaction_id) { -// Get pointer to 1st virtual processor info struct in SRAM - vcpu_t *virtual_processor_table = (vcpu_t*) SV_VCPU; +//! \brief sends the SDP message built in the ::my_msg global +static inline void send_sdp_message(sdp_msg_pure_data *my_msg, uint n_data_words) { + my_msg->tag = tag; // IPTag 1 + my_msg->dest_port = PORT_ETH; // Ethernet + my_msg->dest_addr = sv->eth_addr; // Nearest Ethernet chip - // Get the address this core's DTCM data starts at from the user data - // member of the structure associated with this virtual processor - virtual_processor_table[spin1_get_core_id()].user1 = transaction_id; -} + // fill in SDP source & flag fields + my_msg->flags = 0x07; + my_msg->srce_port = 3; + my_msg->srce_addr = sv->p2p_addr; + my_msg->length = sizeof(sdp_hdr_t) + sizeof(uint) * n_data_words; + if (my_msg->length > ABSOLUTE_MAX_SIZE_OF_SDP_IN_BYTES) { + log_error("bad message length %u", my_msg->length); + } -//! \brief sends the SDP message built in the ::my_msg global -static inline void send_sdp_message(void) { - log_debug("sending message of length %u", my_msg.length); - while (!spin1_send_sdp_msg((sdp_msg_t *) &my_msg, SDP_TIMEOUT)) { + log_debug("sending message of length %u", my_msg->length); + while (!spin1_send_sdp_msg((sdp_msg_t *) my_msg, SDP_TIMEOUT)) { log_debug("failed to send SDP message"); spin1_delay_us(MESSAGE_DELAY_TIME_WHEN_FAIL); } @@ -335,33 +237,16 @@ static inline void send_sdp_message(void) { //! \brief sends a multicast (with payload) message to the current target chip //! \param[in] command: the key offset, which indicates the command being sent //! \param[in] payload: the argument to the command -static inline void send_mc_message(key_offsets command, uint payload) { - uint key = data_in_mc_key_map[chip_x][chip_y] + command; +//! \param[in] key_x: The key x chip to use +//! \param[in] key_y: The key y chip to use +static inline void send_mc_message(key_offsets command, uint payload, + uint key_x, uint key_y) { + uint key = data_in_mc_key_map[key_x][key_y] + command; while (spin1_send_mc_packet(key, payload, WITH_PAYLOAD) == 0) { spin1_delay_us(MESSAGE_DELAY_TIME_WHEN_FAIL); } } -//! \brief Sanity checking for writes, ensuring that they're to the _buffered_ -//! SDRAM range. -//! -//! Note that the RTE here is good as it is better (easier to debug, easier to -//! comprehend) than having corrupt memory actually written. -//! -//! \param[in] write_address: where we are going to write. -//! \param[in] n_elements: the number of words we are going to write. -static inline void sanity_check_write(uint write_address, uint n_elements) { - // determine size of data to send - log_debug("Writing %u elements to 0x%08x", n_elements, write_address); - - uint end_ptr = write_address + n_elements * sizeof(uint); - if (write_address < SDRAM_BASE_BUF || end_ptr >= SDRAM_BASE_UNBUF || - end_ptr < write_address) { - log_error("bad write range 0x%08x-0x%08x", write_address, end_ptr); - rt_error(RTE_SWERR); - } -} - //! \brief sends multicast messages accordingly for an SDP message //! \param[in] data: the actual data from the SDP message //! \param[in] n_elements: the number of data items in the SDP message @@ -371,246 +256,16 @@ static inline void sanity_check_write(uint write_address, uint n_elements) { //! on-chip network overhead //! \param[in] write_address: the sdram address where this block of data is //! to be written to +//! \param[in] key_x: The key x chip to use +//! \param[in] key_y: The key y chip to use static void process_sdp_message_into_mc_messages( - const uint *data, uint n_elements, bool set_write_address, - uint write_address) { - // send mc message with SDRAM location to correct chip - if (set_write_address) { - send_mc_message(WRITE_ADDR_KEY_OFFSET, write_address); - } + const uint *data, uint n_elements, uint key_x, uint key_y) { // send mc messages containing rest of sdp data for (uint data_index = 0; data_index < n_elements; data_index++) { log_debug("data is %d", data[data_index]); - send_mc_message(DATA_KEY_OFFSET, data[data_index]); - } -} - -//! \brief creates a store for sequence numbers in a memory store. -//! -//! May allocate in either DTCM (preferred) or SDRAM. -//! -//! \param[in] max_seq: the max seq num expected during this stage -static void create_sequence_number_bitfield(uint max_seq) { - if (received_seq_nums_store != NULL) { - log_error("Allocating seq num store when already one exists at 0x%08x", - received_seq_nums_store); - rt_error(RTE_SWERR); - } - size_of_bitfield = get_bit_field_size(max_seq + 1); - if (max_seq_num != max_seq) { - max_seq_num = max_seq; - alloc_in_sdram = false; - if (max_seq_num >= SDRAM_VS_DTCM_THRESHOLD || (NULL == - (received_seq_nums_store = spin1_malloc( - size_of_bitfield * sizeof(uint32_t))))) { - received_seq_nums_store = sark_xalloc( - sv->sdram_heap, size_of_bitfield * sizeof(uint32_t), 0, - ALLOC_LOCK | ALLOC_ID | (sark_vec->app_id << 8)); - if (received_seq_nums_store == NULL) { - log_error( - "Failed to allocate %u bytes for missing seq num store", - size_of_bitfield * sizeof(uint32_t)); - rt_error(RTE_SWERR); - } - alloc_in_sdram = true; - } - } - log_debug("clearing bit field"); - clear_bit_field(received_seq_nums_store, size_of_bitfield); -} - -//! \brief Frees the allocated sequence number store. -static inline void free_sequence_number_bitfield(void) { - if (received_seq_nums_store == NULL) { - log_error("Freeing a non-existent seq num store"); - rt_error(RTE_SWERR); - } - if (alloc_in_sdram) { - sark_xfree(sv->sdram_heap, received_seq_nums_store, - ALLOC_LOCK | ALLOC_ID | (sark_vec->app_id << 8)); - } else { - sark_free(received_seq_nums_store); - } - received_seq_nums_store = NULL; - max_seq_num = 0xFFFFFFFF; -} - -//! \brief calculates the new sdram location for a given seq num -//! \param[in] seq_num: the seq num to figure offset for -//! \return the new sdram location. -static inline uint calculate_sdram_address_from_seq_num(uint seq_num) { - return (start_sdram_address - + (DATA_IN_NORMAL_PACKET_WORDS * seq_num * sizeof(uint))); -} - -//! \brief Sets the length of the outbound SDP message in ::my_msg -//! \param[in] end: Points to the first byte after the content of the message -static inline void set_message_length(const void *end) { - my_msg.length = ((const uint8_t *) end) - &my_msg.flags; - if (my_msg.length > ABSOLUTE_MAX_SIZE_OF_SDP_IN_BYTES) { - log_error("bad message length %u", my_msg.length); - } -} - -//! \brief handles reading the address, chips and max packets from a -//! SDP message (command: ::SDP_SEND_DATA_TO_LOCATION_CMD) -//! \param[in] receive_data_cmd: The message to parse -static void process_address_data( - const receive_data_to_location_msg_t *receive_data_cmd) { - // if received when doing a stream. ignore as either clone or oddness - if (received_seq_nums_store != NULL) { - log_debug( - "received location message with transaction id %d when " - "already processing stream with transaction id %d", - receive_data_cmd->transaction_id, transaction_id); - return; - } - - // updater transaction id if it hits the cap - if (((transaction_id + 1) & TRANSACTION_CAP) == 0) { - transaction_id = 0; - publish_transaction_id_to_user_1(transaction_id); - } - - // if transaction id is not as expected. ignore it as its from the past. - // and worthless - if (receive_data_cmd->transaction_id != transaction_id + 1) { - log_debug( - "received location message with unexpected " - "transaction id %d; mine is %d", - receive_data_cmd->transaction_id, transaction_id + 1); - return; - } - - //extract transaction id and update - transaction_id = receive_data_cmd->transaction_id; - publish_transaction_id_to_user_1(transaction_id); - - // track changes - uint prev_x = chip_x, prev_y = chip_y; - - // update sdram and tracker as now have the sdram and size - chip_x = receive_data_cmd->chip_x; - chip_y = receive_data_cmd->chip_y; - - if (prev_x != chip_x || prev_y != chip_y) { - log_debug( - "Changed stream target chip to %d,%d for transaction id %d", - chip_x, chip_y, transaction_id); - } - - log_debug("Writing %u packets to 0x%08x for transaction id %d", - receive_data_cmd->max_seq_num + 1, receive_data_cmd->address, - transaction_id); - - // store where the sdram started, for out-of-order UDP packets. - start_sdram_address = (uint) receive_data_cmd->address; - - // allocate location for holding the seq numbers - create_sequence_number_bitfield(receive_data_cmd->max_seq_num); - total_received_seq_nums = 0; - prov.n_in_streams++; - sdram_prov->n_in_streams = prov.n_in_streams; - - // set start of last seq number - last_seen_seq_num = 0; -} - -//! \brief sends the finished request -static void send_finished_response(void) { - // send boundary key, so that monitor knows everything in the previous - // stream is done - sdp_msg_out_payload_t *payload = (sdp_msg_out_payload_t *) my_msg.data; - send_mc_message(BOUNDARY_KEY_OFFSET, 0); - payload->command = SDP_SEND_FINISHED_DATA_IN_CMD; - my_msg.length = sizeof(sdp_hdr_t) + - sizeof(int) * SEND_MISSING_SEQ_HEADER_WORDS; - send_sdp_message(); - log_debug("Sent end flag"); -} - -//! \brief searches through received sequence numbers and transmits missing ones -//! back to host for retransmission -//! \param[in] msg: The message asking for the missed seq nums -static void process_missing_seq_nums_and_request_retransmission( - const sdp_msg_pure_data *msg) { - // verify in right state - uint this_message_transaction_id = msg->data[TRANSACTION_ID]; - if (received_seq_nums_store == NULL && - this_message_transaction_id != transaction_id) { - log_debug( - "received missing seq numbers before a location with a " - "transaction id which is stale."); - return; - } - if (received_seq_nums_store == NULL && - this_message_transaction_id == transaction_id) { - log_debug("received tell request when already sent finish. resending"); - send_finished_response(); - return; - } - - sdp_msg_out_payload_t *payload = (sdp_msg_out_payload_t *) my_msg.data; - payload->transaction_id = transaction_id; - - // check that missing seq transmission is actually needed, or - // have we finished - if (total_received_seq_nums == max_seq_num + 1) { - free_sequence_number_bitfield(); - total_received_seq_nums = 0; - send_finished_response(); - return; - } - - // sending missing seq nums - log_debug("Looking for %d missing packets", - ((int) max_seq_num + 1) - ((int) total_received_seq_nums)); - payload->command = SDP_SEND_MISSING_SEQ_DATA_IN_CMD; - const uint *data_start = payload->data; - const uint *end_of_buffer = (uint *) (payload + 1); - uint *data_ptr = payload->data; - - // handle case of all missing - if (total_received_seq_nums == 0) { - // send response - data_ptr = payload->data; - *(data_ptr++) = ALL_MISSING_FLAG; - set_message_length(data_ptr); - send_sdp_message(); - return; + send_mc_message(DATA_KEY_OFFSET, data[data_index], key_x, key_y); } - - // handle a random number of missing seqs - for (uint bit = 0; bit <= max_seq_num; bit++) { - if (bit_field_test(received_seq_nums_store, bit)) { - continue; - } - - *(data_ptr++) = bit; - if (data_ptr >= end_of_buffer) { - set_message_length(data_ptr); - send_sdp_message(); - data_ptr = payload->data; - } - } - - // send final message if required - if (data_ptr > data_start) { - set_message_length(data_ptr); - send_sdp_message(); - } -} - -//! \brief Calculates the number of words of data in an SDP message. -//! \param[in] msg: the SDP message, as received from SARK -//! \param[in] data_start: where in the message the data actually starts -//! \return the number of data words in the message -static inline uint n_elements_in_msg( - const sdp_msg_pure_data *msg, const uint *data_start) { - // Offset in bytes from the start of the SDP message to where the data is - uint offset = ((uint8_t *) data_start) - &msg->flags; - return (msg->length - offset) / sizeof(uint); } //! \brief because spin1_memcpy() is stupid, especially for access to SDRAM @@ -626,85 +281,6 @@ static inline void copy_data( } } -//! \brief Handles receipt and parsing of a message full of sequence numbers -//! that need to be retransmitted (command: ::SDP_SEND_SEQ_DATA_CMD) -//! \param[in] msg: The message to parse (really of type receive_seq_data_msg_t) -static inline void receive_seq_data(const sdp_msg_pure_data *msg) { - // cast to the receive seq data - const receive_seq_data_msg_t *receive_data_cmd = - (receive_seq_data_msg_t *) msg->data; - - // check for bad states - if (received_seq_nums_store == NULL) { - log_debug("received data before being given a location"); - return; - } - if (receive_data_cmd->transaction_id != transaction_id) { - log_debug("received data from a different transaction"); - return; - } - - // all good, process data - uint seq = receive_data_cmd->seq_num; - log_debug("Sequence data, seq:%u", seq); - if (seq > max_seq_num) { - log_error("Bad sequence number %u when max is %u!", seq, max_seq_num); - return; - } - - uint this_sdram_address = calculate_sdram_address_from_seq_num(seq); - bool send_sdram_address = (last_seen_seq_num != seq - 1); - - if (!bit_field_test(received_seq_nums_store, seq)) { - bit_field_set(received_seq_nums_store, seq); - total_received_seq_nums++; - } - last_seen_seq_num = seq; - - uint n_elements = n_elements_in_msg(msg, receive_data_cmd->data); - log_debug("n elements is %d", n_elements); - sanity_check_write(this_sdram_address, n_elements); - if (chip_x == 0 && chip_y == 0) { - // directly write the data to where it belongs - for (uint data_index = 0; data_index < n_elements; data_index++) { - log_debug("data is %x", receive_data_cmd->data[data_index]); - } - copy_data( - (address_t) this_sdram_address, receive_data_cmd->data, - n_elements); - } else { - // transmit data to chip; the data lasts to the end of the message - process_sdp_message_into_mc_messages( - receive_data_cmd->data, n_elements, - send_sdram_address, this_sdram_address); - } -} - -//! \brief processes SDP messages for the Data In protocol -//! \param[in] msg: the SDP message -static void data_in_receive_sdp_data(const sdp_msg_pure_data *msg) { - uint command = msg->data[COMMAND_ID]; - prov.n_sdp_recvd++; - sdram_prov->n_sdp_recvd = prov.n_sdp_recvd; - - // check for separate commands - switch (command) { - case SDP_SEND_DATA_TO_LOCATION_CMD: - // translate elements to variables - process_address_data((receive_data_to_location_msg_t *) msg->data); - break; - case SDP_SEND_SEQ_DATA_CMD: - receive_seq_data(msg); - break; - case SDP_TELL_MISSING_BACK_TO_HOST: - log_debug("Checking for missing"); - process_missing_seq_nums_and_request_retransmission(msg); - break; - default: - log_error("Failed to recognise command id %u", command); - } -} - //! \brief sends the basic timeout command via multicast to the extra monitors //! \param[in,out] msg: the request to send the timeout; will be updated with //! result @@ -768,6 +344,82 @@ static void reinjection_sdp_command(sdp_msg_t *msg) { } } +static void send_msg(sdp_msg_t *msg) { + while (!spin1_send_sdp_msg(msg, SDP_TIMEOUT)) { + log_debug("failed to send SDP message"); + spin1_delay_us(MESSAGE_DELAY_TIME_WHEN_FAIL); + } +} + +static uint last_sequence = 0xFFFFFFF0; + +static bool send_in_progress = false; + +static void send_data_over_multicast(sdp_msg_t *msg) { + if (send_in_progress) { + if (last_sequence != msg->seq) { + msg->cmd_rc = RC_BUF; + reflect_sdp_message(msg, 0); + send_msg(msg); + } + return; + } else if (last_sequence == msg->seq) { + msg->cmd_rc = RC_OK; + reflect_sdp_message(msg, 0); + send_msg(msg); + return; + } + + last_sequence = msg->seq; + send_in_progress = true; + + uint *data = &(msg->arg1); + uint length = (msg->length - 12) >> 2; + + while (length > 0) { + + // Read a header + uint address = data[0]; + uint chip_x = (data[1] >> 16) & 0xFFFF; + uint chip_y = data[1] & 0xFFFF; + uint n_data_items = data[2]; + data = &(data[3]); + length -= 3; + + if (chip_x >= 8 || chip_y >= 8) { + log_error("Chip %u, %u is not valid!", chip_x, chip_y); + msg->cmd_rc = RC_ARG; + reflect_sdp_message(msg, 0); + send_msg(msg); + return; + } + + if (n_data_items > length) { + log_error("Not enough data to read %u words from %u remaining", + n_data_items, length); + msg->cmd_rc = RC_ARG; + reflect_sdp_message(msg, 0); + send_msg(msg); + return; + } + + log_debug("Writing using %u words to %u, %u: 0x%08x", n_data_items, + chip_x, chip_y, address); + send_mc_message(WRITE_ADDR_KEY_OFFSET, address, chip_x, chip_y); + process_sdp_message_into_mc_messages(data, n_data_items, + chip_x, chip_y); + + data = &(data[n_data_items]); + length -= n_data_items; + } + send_in_progress = false; + + // set message to correct format + msg->cmd_rc = RC_OK; + reflect_sdp_message(msg, 0); + send_msg(msg); +} + //! \brief processes SDP messages //! \param[in,out] mailbox: the SDP message; will be _freed_ by this call! //! \param[in] port: the port associated with this SDP message @@ -776,8 +428,8 @@ static void receive_sdp_message(uint mailbox, uint port) { case REINJECTION_PORT: reinjection_sdp_command((sdp_msg_t *) mailbox); break; - case DATA_SPEED_UP_IN_PORT: - data_in_receive_sdp_data((sdp_msg_pure_data *) mailbox); + case DATA_COPY_IN_PORT: + send_data_over_multicast((sdp_msg_t *) mailbox); break; default: log_info("unexpected port %d\n", port); @@ -788,15 +440,15 @@ static void receive_sdp_message(uint mailbox, uint port) { //! \brief sends data to the host via SDP (using ::my_msg) static void send_data(void) { + sdp_msg_pure_data my_msg; copy_data(&my_msg.data, data, position_in_store); - my_msg.length = sizeof(sdp_hdr_t) + position_in_store * sizeof(uint); if (seq_num > max_seq_num) { log_error("Got a funky seq num in sending; max is %d, received %d", max_seq_num, seq_num); } - send_sdp_message(); + send_sdp_message(&my_msg, position_in_store); seq_num++; data[SEQ_NUM_LOC] = seq_num; @@ -882,6 +534,7 @@ static void initialise(void) { transaction_id_key = config->transaction_id_key; end_flag_key = config->end_flag_key; basic_data_key = config->basic_data_key; + tag = config->tag_id; log_info("new seq key = %d, first data key = %d, transaction id key = %d, " "end flag key = %d, basic_data_key = %d", @@ -889,14 +542,6 @@ static void initialise(void) { end_flag_key, basic_data_key); log_info("the tag id being used is %d", config->tag_id); - my_msg.tag = config->tag_id; // IPTag 1 - my_msg.dest_port = PORT_ETH; // Ethernet - my_msg.dest_addr = sv->eth_addr; // Nearest Ethernet chip - - // fill in SDP source & flag fields - my_msg.flags = 0x07; - my_msg.srce_port = 3; - my_msg.srce_addr = sv->p2p_addr; // Set up provenance sdram_prov = data_specification_get_region(PROVENANCE_REGION, ds_regions); @@ -925,9 +570,6 @@ static void initialise(void) { // set sdp callback spin1_callback_on(SDP_PACKET_RX, receive_sdp_message, SDP); - - // load user 1 in case this is a consecutive load - publish_transaction_id_to_user_1(transaction_id); } //! \brief This function is called at application start-up. diff --git a/c_common/models/system_models/src/extra_monitor_support.c b/c_common/models/system_models/src/extra_monitor_support.c index 65a3864e5b..0db137f47c 100644 --- a/c_common/models/system_models/src/extra_monitor_support.c +++ b/c_common/models/system_models/src/extra_monitor_support.c @@ -1185,6 +1185,9 @@ static inline void data_in_process_address(uint data) { data_in_first_write_address = data_in_write_address = (address_t) data; } +static const uint SDRAM_TOP_UNBUF = SDRAM_BASE_UNBUF + SDRAM_SIZE; +static const uint SDRAM_TOP_BUF = SDRAM_TOP_UNBUF + SDRAM_SIZE; + //! \brief Writes a word in a stream and advances the write pointer. //! \param[in] data: The word to write static inline void data_in_process_data(uint data) { @@ -1194,6 +1197,12 @@ static inline void data_in_process_data(uint data) { io_printf(IO_BUF, "[ERROR] Write address not set when write data received!\n"); rt_error(RTE_SWERR); } + uint data_addr = (uint) data_in_write_address; + if (((data_addr < SDRAM_BASE_BUF) || (data_addr >= SDRAM_TOP_BUF)) + && ((data_addr < SDRAM_BASE_UNBUF) || (data_addr >= SDRAM_TOP_UNBUF))) { + io_printf(IO_BUF, "[ERROR] Write address 0x%08x is outside SDRAM\n", data_addr); + rt_error(RTE_SWERR); + } *data_in_write_address = data; data_in_write_address++; } diff --git a/spinn_front_end_common/interface/interface_functions/load_data_specification.py b/spinn_front_end_common/interface/interface_functions/load_data_specification.py index 9709c2dcd6..406c5c44ef 100644 --- a/spinn_front_end_common/interface/interface_functions/load_data_specification.py +++ b/spinn_front_end_common/interface/interface_functions/load_data_specification.py @@ -13,10 +13,10 @@ # limitations under the License. import logging +from functools import partial from typing import Any, Callable - -import numpy from typing_extensions import TypeAlias +import numpy from spinn_utilities.config_holder import get_config_bool from spinn_utilities.progress_bar import ProgressBar @@ -32,6 +32,9 @@ from spinn_front_end_common.utilities.emergency_recovery import ( emergency_recover_states_from_failure) from spinn_front_end_common.interface.ds import DsSqlliteDatabase +from spinn_front_end_common.utilities.iobuf_extractor import IOBufExtractor +from spinn_front_end_common.utilities.scp import ( + WriteMemoryUsingMulticastProcess) logger = FormatAdapter(logging.getLogger(__name__)) _Writer: TypeAlias = Callable[[int, int, int, bytes], Any] @@ -118,6 +121,12 @@ def load_data_specs(self, is_system: bool, uses_advanced_monitors: bool): self.__java_app(uses_advanced_monitors) else: self.__python_load(is_system, uses_advanced_monitors) + if uses_advanced_monitors and get_config_bool( + "Reports", "write_advanced_monitor_iobuf"): + extractor = IOBufExtractor( + None, recovery_mode=True, + filename_template="system_iobuf_{}_{}_{}.txt") + extractor.extract_iobuf() except: # noqa: E722 if uses_advanced_monitors: emergency_recover_states_from_failure() @@ -140,6 +149,8 @@ def __python_load(self, is_system: bool, uses_advanced_monitors: bool): """ if uses_advanced_monitors: self.__set_router_timeouts() + write_memory_process = WriteMemoryUsingMulticastProcess( + FecDataView.get_scamp_connection_selector()) # create a progress bar for end users with DsSqlliteDatabase() as ds_database: @@ -165,7 +176,11 @@ def __python_load(self, is_system: bool, uses_advanced_monitors: bool): for x, y, p, eth_x, eth_y in progress.over(core_infos): if uses_advanced_monitors: gatherer = FecDataView.get_gatherer_by_xy(eth_x, eth_y) - writer = gatherer.send_data_into_spinnaker + placement = FecDataView.get_placement_of_vertex(gatherer) + writer = partial( + write_memory_process.write_memory_from_bytearray, + placement.x, placement.y, placement.p) + written = self.__python_load_core(ds_database, x, y, p, writer) to_write = ds_database.get_memory_to_write(x, y, p) if written != to_write: diff --git a/spinn_front_end_common/interface/java_caller.py b/spinn_front_end_common/interface/java_caller.py index e2ffab27fe..77d2c12e91 100644 --- a/spinn_front_end_common/interface/java_caller.py +++ b/spinn_front_end_common/interface/java_caller.py @@ -424,7 +424,7 @@ def load_app_data_specification(self, use_monitors: bool) -> None: """ if use_monitors: result = self._run_java( - 'dse_app_mon', self._placement_json, self._machine_json(), + 'dse_app_mon_mc', self._placement_json, self._machine_json(), FecDataView.get_ds_database_path(), FecDataView.get_run_dir_path()) else: diff --git a/spinn_front_end_common/interface/spinnaker.cfg b/spinn_front_end_common/interface/spinnaker.cfg index 1f305ff7fd..c75d60889a 100644 --- a/spinn_front_end_common/interface/spinnaker.cfg +++ b/spinn_front_end_common/interface/spinnaker.cfg @@ -45,6 +45,7 @@ write_bit_field_compressor_report = False write_drift_report_start = False write_drift_report_end = False write_text_specs = False +write_advanced_monitor_iobuf = False max_reports_kept = 10 remove_errored_folders = False diff --git a/spinn_front_end_common/utilities/scp/__init__.py b/spinn_front_end_common/utilities/scp/__init__.py index 0857ea6303..bbf8532dd1 100644 --- a/spinn_front_end_common/utilities/scp/__init__.py +++ b/spinn_front_end_common/utilities/scp/__init__.py @@ -16,9 +16,12 @@ from .load_mc_routes_process import LoadMCRoutesProcess from .reinjector_control_process import ReinjectorControlProcess from .update_runtime_process import UpdateRuntimeProcess +from .write_memory_using_multicast_process import ( + WriteMemoryUsingMulticastProcess) __all__ = ( "ClearIOBUFProcess", "LoadMCRoutesProcess", "ReinjectorControlProcess", - "UpdateRuntimeProcess") + "UpdateRuntimeProcess", + "WriteMemoryUsingMulticastProcess") diff --git a/spinn_front_end_common/utilities/scp/write_memory_using_multicast_process.py b/spinn_front_end_common/utilities/scp/write_memory_using_multicast_process.py new file mode 100644 index 0000000000..15ba0ec740 --- /dev/null +++ b/spinn_front_end_common/utilities/scp/write_memory_using_multicast_process.py @@ -0,0 +1,84 @@ +# Copyright (c) 2017 The University of Manchester +# +# 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 +# +# https://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. +from typing import Optional, Tuple +import numpy +from numpy import uint8, uint32 +from spinnman.constants import UDP_MESSAGE_MAX_SIZE +from spinnman.processes import AbstractMultiConnectionProcess +from spinnman.messages.scp.impl import CheckOKResponse +from spinn_front_end_common.utilities.utility_objs.\ + extra_monitor_scp_messages import SendMCDataMessage + +_UNSIGNED_WORD = 0xFFFFFFFF + + +class WriteMemoryUsingMulticastProcess( + AbstractMultiConnectionProcess[CheckOKResponse]): + """ + How to send messages to write data using multicast + """ + __slots__ = () + + def write_memory_from_bytearray( + self, x: int, y: int, p: int, target_x: int, target_y: int, + base_address: int, data: bytes, data_offset: int = 0, + n_bytes: Optional[int] = None, + get_sum: bool = False) -> Tuple[int, int]: + """ + Write memory using multicast over the advanced monitor connection. + + :param int x: + The x-coordinate of the chip where the advanced monitor is + :param int y: + The y-coordinate of the chip where the advanced monitor is + :param int p: + The processor ID of the chip where the advanced monitor is + :param int target_x: + The x-coordinate of the chip where the memory is to be written to + :param int target_y: + The y-coordinate of the chip where the memory is to be written to + :param int base_address: + The address in SDRAM where the region of memory is to be written + :param bytes data: The data to write. + :param int n_bytes: + The amount of data to be written in bytes. If not specified, + length of data is used. + :param int data_offset: The offset from which the valid data begins + :param bool get_sum: whether to return a checksum or 0 + :return: The number of bytes written, the checksum (0 if get_sum=False) + :rtype: int, int + """ + offset = 0 + if n_bytes is None: + n_bytes_to_write = len(data) + else: + n_bytes_to_write = n_bytes + n_bytes_written = n_bytes_to_write + with self._collect_responses(): + while n_bytes_to_write > 0: + bytes_to_send = min(n_bytes_to_write, UDP_MESSAGE_MAX_SIZE) + data_array = data[data_offset:data_offset + bytes_to_send] + + self._send_request( + SendMCDataMessage(x, y, p, target_x, target_y, + base_address + offset, data_array)) + + n_bytes_to_write -= bytes_to_send + offset += bytes_to_send + data_offset += bytes_to_send + if not get_sum: + return n_bytes_written, 0 + np_data = numpy.array(data, dtype=uint8) + np_sum = int(numpy.sum(np_data.view(uint32), dtype=uint32)) + return n_bytes_written, np_sum & _UNSIGNED_WORD diff --git a/spinn_front_end_common/utilities/utility_objs/extra_monitor_scp_messages/__init__.py b/spinn_front_end_common/utilities/utility_objs/extra_monitor_scp_messages/__init__.py index 16f63f294d..6c9650ccc9 100644 --- a/spinn_front_end_common/utilities/utility_objs/extra_monitor_scp_messages/__init__.py +++ b/spinn_front_end_common/utilities/utility_objs/extra_monitor_scp_messages/__init__.py @@ -21,6 +21,7 @@ from .set_reinjection_packet_types_message import ( SetReinjectionPacketTypesMessage) from .set_router_timeout_message import SetRouterTimeoutMessage +from .send_mc_data_message import SendMCDataMessage __all__ = ( "ClearReinjectionQueueMessage", @@ -30,4 +31,5 @@ "LoadSystemMCRoutesMessage", "ResetCountersMessage", "SetReinjectionPacketTypesMessage", - "SetRouterTimeoutMessage") + "SetRouterTimeoutMessage", + "SendMCDataMessage") diff --git a/spinn_front_end_common/utilities/utility_objs/extra_monitor_scp_messages/send_mc_data_message.py b/spinn_front_end_common/utilities/utility_objs/extra_monitor_scp_messages/send_mc_data_message.py new file mode 100644 index 0000000000..3d86357925 --- /dev/null +++ b/spinn_front_end_common/utilities/utility_objs/extra_monitor_scp_messages/send_mc_data_message.py @@ -0,0 +1,60 @@ +# Copyright (c) 2017 The University of Manchester +# +# 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 +# +# https://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. + +from spinn_utilities.overrides import overrides +from spinnman.messages.scp import SCPRequestHeader +from spinnman.messages.scp.abstract_messages import AbstractSCPRequest +from spinnman.messages.sdp import SDPFlag, SDPHeader +from spinnman.messages.scp.impl.check_ok_response import CheckOKResponse +from spinnman.messages.scp.enums import SCPCommand +from spinnman.model.enums import SDP_PORTS +from spinn_front_end_common.utilities.constants import BYTES_PER_WORD + +X_SHIFT = 16 + + +class SendMCDataMessage(AbstractSCPRequest[CheckOKResponse]): + """ + An SCP Request to write data using multicast. + """ + + __slots__ = () + + def __init__(self, x: int, y: int, p: int, target_x: int, target_y: int, + target_address: int, data: bytes): + """ + :param int x: The x-coordinate of a chip, between 0 and 255 + :param int y: The y-coordinate of a chip, between 0 and 255 + :param int p: + The processor running the extra monitor vertex, between 0 and 17 + """ + super().__init__( + SDPHeader( + flags=SDPFlag.REPLY_EXPECTED, + destination_port=( + SDP_PORTS.EXTRA_MONITOR_CORE_COPY_DATA_IN.value), + destination_cpu=p, destination_chip_x=x, + destination_chip_y=y), + SCPRequestHeader( + command=SCPCommand.CMD_WRITE), + argument_1=target_address, + argument_2=target_x << X_SHIFT | target_y, + argument_3=len(data) // BYTES_PER_WORD, + data=data) + + @overrides(AbstractSCPRequest.get_scp_response) + def get_scp_response(self) -> CheckOKResponse: + return CheckOKResponse( + "write_over_multicast", + SCPCommand.CMD_WRITE) diff --git a/spinn_front_end_common/utility_models/data_speed_up_packet_gatherer_machine_vertex.py b/spinn_front_end_common/utility_models/data_speed_up_packet_gatherer_machine_vertex.py index a4b28e2e0b..2af7f77f32 100644 --- a/spinn_front_end_common/utility_models/data_speed_up_packet_gatherer_machine_vertex.py +++ b/spinn_front_end_common/utility_models/data_speed_up_packet_gatherer_machine_vertex.py @@ -13,26 +13,22 @@ # limitations under the License. from __future__ import annotations import os -import datetime import logging import time import struct from enum import Enum, IntEnum from typing import ( - Any, BinaryIO, Final, Iterable, List, Optional, Set, Tuple, Union, - TYPE_CHECKING) - + Final, Iterable, List, Optional, Set, Tuple, TYPE_CHECKING) from spinn_utilities.config_holder import get_config_bool from spinn_utilities.overrides import overrides from spinn_utilities.log import FormatAdapter from spinn_utilities.typing.coords import XY -from spinn_machine import Chip from spinnman.exceptions import SpinnmanTimeoutException from spinnman.messages.sdp import SDPMessage, SDPHeader, SDPFlag from spinnman.model.enums import ( - CPUState, ExecutableType, SDP_PORTS, UserRegister) + CPUState, ExecutableType, SDP_PORTS) from spinnman.connections.udp_packet_connections import SCAMPConnection from pacman.model.graphs.machine import MachineVertex @@ -157,17 +153,6 @@ class _DataOutCommands(IntEnum): CLEAR = 2000 -class _DataInCommands(IntEnum): - """ - Command IDs for the SDP packets for data in. - """ - SEND_DATA_TO_LOCATION = 200 - SEND_SEQ_DATA = 2000 - SEND_TELL = 2001 - RECEIVE_MISSING_SEQ_DATA = 2002 - RECEIVE_FINISHED = 2003 - - # precompiled structures _ONE_WORD = struct.Struct(" ConstantSDRAM: @@ -349,15 +318,6 @@ def iptags(self) -> List[IPtagResource]: port=self._TAG_INITIAL_PORT, strip_sdp=True, ip_address="localhost", traffic_identifier="DATA_SPEED_UP")] - def _read_transaction_id_from_machine(self) -> None: - """ - Looks up from the machine what the current transaction ID is - and updates the data speed up gatherer. - """ - self._transaction_id = FecDataView.get_transceiver().read_user( - self._placement.x, self._placement.y, self._placement.p, - UserRegister.USER_1) - @overrides(AbstractHasAssociatedBinary.get_binary_start_type) def get_binary_start_type(self) -> ExecutableType: return ExecutableType.SYSTEM @@ -366,8 +326,6 @@ def get_binary_start_type(self) -> ExecutableType: def generate_data_specification( self, spec: DataSpecificationGenerator, placement: Placement): # pylint: disable=unsubscriptable-object - # update my placement for future knowledge - self.__placement = placement # Create the data regions for hello world self._reserve_memory_regions(spec) @@ -424,12 +382,6 @@ def generate_data_specification( # End-of-Spec: spec.end_specification() - @property - def _placement(self) -> Placement: - if self.__placement is None: - raise SpinnFrontEndException("placement not known") - return self.__placement - def _reserve_memory_regions(self, spec: DataSpecificationGenerator): """ Writes the DSG regions memory sizes. Static so that it can be used @@ -453,135 +405,6 @@ def _reserve_memory_regions(self, spec: DataSpecificationGenerator): def get_binary_file_name(self) -> str: return "data_speed_up_packet_gatherer.aplx" - def _generate_data_in_report( - self, time_diff, data_size: int, x: int, y: int, - address_written_to: int, missing_seq_nums): - """ - Writes the data in report for this stage. - - :param ~datetime.timedelta time_diff: - the time taken to write the memory - :param int data_size: the size of data that was written in bytes - :param int x: - the location in machine where the data was written to X axis - :param int y: - the location in machine where the data was written to Y axis - :param int address_written_to: where in SDRAM it was written to - :param list(set(int)) missing_seq_nums: - the set of missing sequence numbers per data transmission attempt - """ - dir_path = FecDataView.get_run_dir_path() - in_report_path = os.path.join(dir_path, self.IN_REPORT_NAME) - if not os.path.isfile(in_report_path): - with open(in_report_path, "w", encoding="utf-8") as writer: - writer.write( - "x\t\t y\t\t SDRAM address\t\t size in bytes\t\t\t" - " time took \t\t\t Mb/s \t\t\t missing sequence numbers\n") - writer.write( - "------------------------------------------------" - "------------------------------------------------" - "-------------------------------------------------\n") - - time_took_ms = float(time_diff.microseconds + - time_diff.total_seconds() * 1000000) - megabits = (data_size * 8.0) / (1024 * BYTES_PER_KB) - if time_took_ms == 0: - mbs: Any = "unknown, below threshold" - else: - mbs = megabits / (float(time_took_ms) / 100000.0) - - with open(in_report_path, "a", encoding="utf-8") as writer: - writer.write( - f"{x}\t\t {y}\t\t {address_written_to}\t\t {data_size}\t\t" - f"\t\t {time_took_ms}\t\t\t {mbs}\t\t {missing_seq_nums}\n") - - def send_data_into_spinnaker( - self, x: int, y: int, base_address: int, - data: Union[BinaryIO, bytes, str, int], *, - n_bytes: Optional[int] = None, offset: int = 0, - cpu: int = 0): # pylint: disable=unused-argument - """ - Sends a block of data into SpiNNaker to a given chip. - - :param int x: chip x for data - :param int y: chip y for data - :param int base_address: the address in SDRAM to start writing memory - :param data: - the data to write or filename to load data from (if a string) - :type data: bytes or bytearray or memoryview or str - :param int n_bytes: how many bytes to read, or `None` if not set - :param int offset: where in the data to start from - :param int cpu: Ignored; can only target SDRAM so unimportant - """ - # if file, read in and then process as normal - if isinstance(data, str): - if offset != 0: - raise ValueError( - "when using a file, you can only have a offset of 0") - - with open(data, "rb") as reader: - # n_bytes=None already means 'read everything' - data = reader.read(n_bytes) - # Number of bytes to write is now length of buffer we have - if n_bytes is None: - n_bytes = len(data) - else: - n_bytes = min(n_bytes, len(data)) - elif not isinstance(data, (bytes, bytearray)): - raise ValueError("that type of data not supported") - if n_bytes is None: - n_bytes = len(data) - if n_bytes < 0: - raise ValueError("cannot write a negative amount of data") - - destination = FecDataView.get_chip_at(x, y) - # start time recording - start = datetime.datetime.now() - # send data - self._send_data_via_extra_monitors( - destination, base_address, data[offset:n_bytes + offset]) - # end time recording - end = datetime.datetime.now() - - if VERIFY_SENT_DATA: - original_data = bytes(data[offset:n_bytes + offset]) - transceiver = FecDataView.get_transceiver() - verified_data = bytes(transceiver.read_memory( - x, y, base_address, n_bytes)) - self.__verify_sent_data( - original_data, verified_data, x, y, base_address, n_bytes) - - # write report - if get_config_bool("Reports", "write_data_speed_up_reports"): - self._generate_data_in_report( - x=x, y=y, time_diff=end - start, - data_size=n_bytes, address_written_to=base_address, - missing_seq_nums=self._missing_seq_nums_data_in) - - @staticmethod - def __verify_sent_data( - original_data: bytes, verified_data: bytes, x: int, y: int, - base_address: int, n_bytes: int): - if original_data != verified_data: - log.error("VARIANCE: chip:{},{} address:{} len:{}", - x, y, base_address, n_bytes) - log.error("original:{}", original_data.hex()) - log.error("verified:{}", verified_data.hex()) - for i, (a, b) in enumerate(zip(original_data, verified_data)): - if a != b: - raise ValueError(f"Mismatch found as position {i}") - - def __make_data_in_message(self, payload: bytes) -> SDPMessage: - return SDPMessage( - sdp_header=SDPHeader( - destination_chip_x=self._placement.x, - destination_chip_y=self._placement.y, - destination_cpu=self._placement.p, - destination_port=( - SDP_PORTS.EXTRA_MONITOR_CORE_DATA_IN_SPEED_UP.value), - flags=SDPFlag.REPLY_NOT_EXPECTED), - data=payload) - @staticmethod def __make_data_out_message( placement: Placement, payload: bytes) -> SDPMessage: @@ -607,270 +430,6 @@ def __open_connection(self) -> SCAMPConnection: retarget_tag(connection, self._x, self._y, self._remote_tag) return connection - def _send_data_via_extra_monitors( - self, destination_chip: Chip, start_address: int, - data_to_write: bytes): - """ - Sends data using the extra monitor cores. - - :param Chip destination_chip: chip to send to - :param int start_address: start address in SDRAM to write data to - :param bytearray data_to_write: the data to write - """ - # Set up the connection - with self.__open_connection() as connection: - # how many packets after first one we need to send - self._max_seq_num = ceildiv( - len(data_to_write), BYTES_IN_FULL_PACKET_WITH_KEY) - - # determine board chip IDs, as the LPG does not know - # machine scope IDs - machine = FecDataView.get_machine() - dest_x, dest_y = machine.get_local_xy(destination_chip) - self._coord_word = (dest_x << DEST_X_SHIFT) | dest_y - - # for safety, check the transaction id from the machine before - # updating - self._read_transaction_id_from_machine() - self._transaction_id = ( - self._transaction_id + 1) & TRANSACTION_ID_CAP - time_out_count = 0 - - # verify completed - received_confirmation = False - while not received_confirmation: - # send initial attempt at sending all the data - self._send_all_data_based_packets( - data_to_write, start_address, connection) - - # Don't create a missing buffer until at least one packet has - # come back. - missing: Optional[Set[int]] = None - - while not received_confirmation: - try: - # try to receive a confirmation of some sort from - # spinnaker - data = connection.receive( - timeout=self._TIMEOUT_PER_RECEIVE_IN_SECONDS) - time_out_count = 0 - - # Read command and transaction id - (cmd, transaction_id) = _TWO_WORDS.unpack_from(data, 0) - - # If wrong transaction id, ignore packet - if self._transaction_id != transaction_id: - continue - - # Decide what to do with the packet - if cmd == _DataInCommands.RECEIVE_FINISHED: - received_confirmation = True - break - - if cmd != _DataInCommands.RECEIVE_MISSING_SEQ_DATA: - raise ValueError(f"Unknown command {cmd} received") - - # The currently received packet has missing sequence - # numbers. Accumulate and dispatch transactionId when - # we've got them all. - if missing is None: - missing = set() - self._missing_seq_nums_data_in.append(missing) - seen_last, seen_all = self._read_in_missing_seq_nums( - data, - BYTES_FOR_RECEPTION_COMMAND_AND_ADDRESS_HEADER, - missing) - - # Check that you've seen something that implies ready - # to retransmit. - if seen_all or seen_last: - self._outgoing_retransmit_missing_seq_nums( - data_to_write, missing, connection) - missing.clear() - - except SpinnmanTimeoutException as e: - # if the timeout has not occurred x times, keep trying - time_out_count += 1 - if time_out_count > TIMEOUT_RETRY_LIMIT: - emergency_recover_state_from_failure( - self, self._placement) - raise SpinnFrontEndException( - "Failed to hear from the machine during " - f"{time_out_count} attempts. " - "Please try removing firewalls.") from e - - # If we never received a packet, we will never have - # created the buffer, so send everything again - if missing is None: - break - - self._outgoing_retransmit_missing_seq_nums( - data_to_write, missing, connection) - missing.clear() - - def _read_in_missing_seq_nums( - self, data: bytes, position: int, - seq_nums: Set[int]) -> Tuple[bool, bool]: - """ - Handles a missing sequence number packet from SpiNNaker. - - :param data: the data to translate into missing sequence numbers - :type data: bytearray or bytes - :param int position: the position in the data to write. - :param set(int) seq_nums: a set of sequence numbers to add to - :return: seen_last flag and seen_all flag - :rtype: tuple(bool, bool) - """ - # find how many elements are in this packet - n_elements = (len(data) - position) // BYTES_PER_WORD - - # store missing - new_seq_nums = n_word_struct(n_elements).unpack_from( - data, position) - - # add missing sequence numbers accordingly - seen_last = False - seen_all = False - if new_seq_nums[-1] == self._MISSING_SEQ_NUMS_END_FLAG: - new_seq_nums = new_seq_nums[:-1] - seen_last = True - if new_seq_nums[-1] == self.FLAG_FOR_MISSING_ALL_SEQUENCES: - for missing_seq in range(self._max_seq_num or 0): - seq_nums.add(missing_seq) - seen_all = True - else: - seq_nums.update(new_seq_nums) - - return seen_last, seen_all - - def _outgoing_retransmit_missing_seq_nums( - self, data_to_write: bytes, missing: Set[int], - connection: SCAMPConnection): - """ - Transmits back into SpiNNaker the missing data based off missing - sequence numbers. - - :param bytearray data_to_write: the data to write. - :param set(int) missing: a set of missing sequence numbers - :type connection: - ~spinnman.connections.udp_packet_connections.SCAMPConnection - """ - missing_seqs_as_list = list(missing) - missing_seqs_as_list.sort() - - # send sequence data - for missing_seq_num in missing_seqs_as_list: - message, _length = self.__make_data_in_stream_message( - data_to_write, missing_seq_num, None) - self.__throttled_send(message, connection) - - # request an update on what is missing - self.__send_tell_flag(connection) - - @staticmethod - def __position_from_seq_number(seq_num: int) -> int: - """ - Calculates where in the raw data to start reading from, given a - sequence number. - - :param int seq_num: the sequence number to determine position from - :return: the position in the byte data - :rtype: int - """ - return BYTES_IN_FULL_PACKET_WITH_KEY * seq_num - - def __make_data_in_stream_message( - self, data_to_write: bytes, seq_num: int, - position: Optional[int]) -> Tuple[SDPMessage, int]: - """ - Determine the data needed to be sent to the SpiNNaker machine - given a sequence number. - - :param bytearray data_to_write: - the data to write to the SpiNNaker machine - :param int seq_num: the sequence number to get the data for - :param position: - the position in the data to write to SpiNNaker, - or None to infer from the sequence number - :type position: int or None - :return: SDP message and how much data has been written - :rtype: tuple(~.SDPMessage, int) - """ - # check for last packet - packet_data_length = BYTES_IN_FULL_PACKET_WITH_KEY - - # determine position in data if not given - if position is None: - position = self.__position_from_seq_number(seq_num) - - # if less than a full packet worth of data, adjust length - if position + packet_data_length > len(data_to_write): - packet_data_length = len(data_to_write) - position - - if packet_data_length < 0: - raise ValueError("weird packet data length") - - # create message body - packet_data = _THREE_WORDS.pack( - _DataInCommands.SEND_SEQ_DATA, self._transaction_id, - seq_num) + data_to_write[position:position+packet_data_length] - - # return message for sending, and the length in data sent - return self.__make_data_in_message(packet_data), packet_data_length - - def __send_location(self, start_address: int, connection: SCAMPConnection): - """ - Send location as separate message. - - :param int start_address: SDRAM location - """ - connection.send_sdp_message(self.__make_data_in_message( - _FIVE_WORDS.pack( - _DataInCommands.SEND_DATA_TO_LOCATION, - self._transaction_id, start_address, self._coord_word, - self._max_seq_num - 1))) - log.debug( - "start address for transaction {} is {}", - self._transaction_id, start_address) - - def __send_tell_flag(self, connection: SCAMPConnection) -> None: - """ - Send tell flag as separate message. - """ - connection.send_sdp_message(self.__make_data_in_message( - _TWO_WORDS.pack( - _DataInCommands.SEND_TELL, self._transaction_id))) - - def _send_all_data_based_packets( - self, data_to_write: bytes, start_address: int, - connection: SCAMPConnection): - """ - Send all the data as one block. - - :param bytearray data_to_write: the data to send - :param int start_address: - """ - # Send the location - self.__send_location(start_address, connection) - - # where in the data we are currently up to - position_in_data = 0 - - # send rest of data - for seq_num in range(self._max_seq_num or 0): - # put in command flag and sequence number - message, length_to_send = self.__make_data_in_stream_message( - data_to_write, seq_num, position_in_data) - position_in_data += length_to_send - - # send the message - self.__throttled_send(message, connection) - log.debug("sent sequence {} of {} bytes", seq_num, length_to_send) - - # check for end flag - self.__send_tell_flag(connection) - log.debug("sent end flag") - def set_cores_for_data_streaming(self) -> None: """ Helper method for setting the router timeouts to a state usable diff --git a/unittests/interface/interface_functions/test_load_data_specification.py b/unittests/interface/interface_functions/test_load_data_specification.py index 7a3ea783ad..894fcc44ae 100644 --- a/unittests/interface/interface_functions/test_load_data_specification.py +++ b/unittests/interface/interface_functions/test_load_data_specification.py @@ -21,6 +21,9 @@ from spinn_machine.version.version_strings import VersionStrings from spinnman.transceiver.version5transceiver import Version5Transceiver from spinnman.model.enums import ExecutableType +from spinnman.processes import MostDirectConnectionSelector +from spinnman.connections.udp_packet_connections import SCAMPConnection +from spinnman.exceptions import SpinnmanTimeoutException from pacman.model.graphs.machine import SimpleMachineVertex from pacman.model.placements import Placements from spinn_front_end_common.abstract_models import AbstractHasAssociatedBinary @@ -36,6 +39,14 @@ from spinn_front_end_common.utilities.exceptions import DataSpecException +class _MockConnection(SCAMPConnection): + def send(self, data): + pass + + def receive_scp_response(self, timeout=1.0): + raise SpinnmanTimeoutException("Test", timeout) + + class _MockTransceiver(Version5Transceiver): """ Pretend transceiver """ @@ -70,6 +81,10 @@ def write_memory( # bogus return for mypy return (-1, -1) + @overrides(Version5Transceiver.get_scamp_connection_selector) + def get_scamp_connection_selector(self) -> MostDirectConnectionSelector: + return MostDirectConnectionSelector([_MockConnection(0, 0)]) + @overrides(Version5Transceiver.close) def close(self) -> None: pass