Idle Hands

Buffers Continued

Since the last time I posted about ring buffers, they've been appearing in several places in my latest project. I started with a hard coded ring buffer in a serial port driver, but I discovered a need for ring buffers in my SPI driver as well. So I refactored the code into its own library header file and saved a few bytes of flash memory in the process. Grab the code and use it in your own projects.

I want to highlight one small point that caught me by surprise when I was writing the library. Take a look at the code that declares the queue in global memory.

/**
 * Declares a circular queue of the specified size as a global 
 * variable.
 */
#define CQ_DECLARE(var_name, size)          \
  static volatile struct _##var_name##_tag  \
  {                                         \
    uint8_t rIdx;                           \
    uint8_t wIdx;                           \
    uint8_t buf[(size)];                    \
  } var_name;

The queue is declared volatile to make sure the compiler won't register optimize accesses to the read and write indexes in case you want to poll the queue's state. This means that every reference to either index results in a load from SRAM. A quick attempt at writing a routine to read a byte might look like the following code.

/**
 * Reads the next byte from the queue into the specified variable. 
 * Caller is responsible for making sure the queue is not empty 
 * before the call.
 */
#define CQ_READ(q, d)                            \
  {                                              \
    (d) = (q).buf[(q).rIdx];                     \
    (q).rIdx = (++(q).rIdx) % sizeof((q).buf);   \
  }

This gets the job done, however, it results in three unnecessary accesses to the read index. The accesses come from the increment operation ++(q).rIdx. Since the queue is declared volatile, the compiler must reload the read index instead of reusing it from the previous line. It will increment the value, store it to SRAM and then reload it again for the modulus operation. Three accesses that could be avoided if the index used for the array lookup were only saved in a register for use later.

Declaring the queue as volatile is necessary for any polling operations to work properly. So, we must introduce another variable that's not so strict with its load and store requirements. Take a look at the code below.

/**
 * Reads the next byte from the queue into the specified variable. 
 * Caller is responsible for making sure the queue is not empty 
 * before the call.
 */
#define CQ_READ(q, d)                       \
  {                                         \
    uint8_t t = (q).rIdx;                   \
    (d) = (q).buf[(t)];                     \
    (q).rIdx = (++t) % sizeof((q).buf);     \
  }

It uses a temporary variable that the compiler can optimize into a single register. For a microcontroller, reading from and writing to SRAM is an expensive operation. The new version saves a few bytes of program memory and a few clock cycles as well. I wouldn't have caught this problem had I not looked at the compiler's assembly output. Lesson learned: when you're writing performance critical code, verify that the CPU is doing what you want for each and every tick of the clock.

Customer Service

My favourite part about embedded development is that at the end of the day, there's always something I can point to and say "I made this." Having something real to hold is pretty satisfying. An added bonus is that you automatically get to pass the girlfriend/grandmother test. Even if they don't know too much about it, the important woman in your life can always point to something tangible when they're asked what you do. Unfortunately, making tiny cool things can get expensive.

The cost of PCB fabrication quickly runs into triple digits even when you simply need to make a few small boards for prototyping a design. So it's great to find a fabrication business that caters to the economical needs of electronics hobbyists. That's why I love working with BatchPCB when I need to take something from the breadboard to a real prototype. They provide a low cost PCB fabrication service and plenty of help for people who are just beginning to dabble in embedded design.

I recently placed an order for two 3 inch square PCBs for a project that I've been working on for a few months. I clicked the purchase button on October 23, and the boards arrived today. Total turnaround time: 15 days. Not only did the boards arrive quickly, they also doubled my order. I found two extra copies of my design in the package. Four boards would have easily pushed my purchase to over $100, so the extra PCBs were a very generous surprise. Thanks goes out to the folks at BatchPCB for their speed and generous gift. I'll definitely be ordering with you again. Now, if only there are no layout errors...

A Work of Art

Antoine de Saint-Exupery is credited for the phrase...

"Perfection is achieved, not when there is nothing more to add, but when there is nothing left to take away."

No clearer have I seen this reflected in modern technology than by what rossum has been able to achieve with an AVR microcontroller and a micro SD flash card. The first Google Android phones were barely able to achieve smooth inertia scrolling, but somehow he's done it with an 8 bit microcontroller running at 12 MHz. A work of art for sure.

Let Freedom Ring

Deleting code. Making things simple. It's one of the most refreshing aspects of software development. The great news is, you get to do a lot of it when you make cool tiny things. Sometimes it's because there isn't any space for complicated designs. But it's most fun when you realize that you didn't need something so intricate in the first place.

Here's a simple way that you can avoid using mutexes in your code. It comes in very handy when doing things in ISR context, since locking a mutex inside an ISR is prohibited in most OSs. Consider the typical fifo. Not a dequeue that uses a linked list, but fixed size ring buffer. Choose your design right and magic happens. We can design a thread-safe data structure without using any OS thread synchronization primitives.

There are three things that need bookkeeping in ring buffers. The read position, the write position and the full/empty flag. Wikipedia has a great discussion of pros and cons of the full/empty buffer debate. In this case we'll choose to take the hit and lose a buffer position to avoid the need for a count. As for the read and write position, they'll be stored separately and explicitly.

The buffer definition looks something like this...

/**
 * The number of bytes to store in the ring buffer.
 */
enum { RING_BUFFER_SIZE = 127 };

/**
 * A ring buffer that stores bytes.
 */
typedef struct _ring_buffer_t
{
  uint8_t rIdx;
  uint8_t wIdx;
  uint8_t buf[RING_BUFFER_SIZE+1];
} ring_buffer_t;

/**
 * Ring buffer in memory. Volatile keyword is a MUST.
 */
static volatile ring_buffer_t rb;

Pay attention to the buffer size. We're using uint8_t's to store the positions, so the buffer can't get larger than 256 bytes. To keep loads and stores atomic, we need to make sure that the positions are represented by types that match the CPU architecture. Use uint8_t for 8-bit microcontrollers and uint32_t for ARM and 32 bit x86 CPUs. Otherwise the CPU will need more than one instruction for the memory access. Interrupts could preempt the operation and retrieve invalid data.

As follows from the design choices above, the write position will always point to an empty space. One byte of the buffer is lost, however it removes the need for a count variable. So the buffer is empty when read and write positions are equal, and full when the read is just ahead of the write. Let's add some test macros...

/**
 * The buffer is empty when the read and write indicies are equal.
 */
#define RING_BUFFER_EMPTY(b) ((b).rIdx == (b).wIdx)

/**
 * The buffer is full when the read index is one entry ahead of the 
 * write index.
 */
#define RING_BUFFER_FULL(b) \
  ((b).rIdx == ((b).wIdx + 1) % sizeof((b).buf))

The behaviour when adding and removing elements is as you'd expect. Whenever something is read, increment the read pointer, whenever something is written, increment the write pointer. The final rules define what happens on full and empty conditions. Trying to write to a full buffer will fail, as will trying to read from an empty buffer. All other operations succeed. The functions might look like this...

/**
 * Reads a byte from the ring buffer.
 */
uint8_t ring_buffer_read(uint8_t* pByte)
{
  /*---------------------------------------------------------------
    Reads on an empty buffer fail.
    -------------------------------------------------------------*/
  if(RING_BUFFER_EMPTY(rb)) { return 1; }

  /*---------------------------------------------------------------
    Grab one byte from the buffer and return it.
    -------------------------------------------------------------*/
  *pByte = rb.buf[rb.rIdx];
  rb.rIdx = (rb.rIdx + 1) % sizeof(rb.buf);
  return 0;
}

/**
 * Writes a byte to the ring buffer.
 */
uint8_t ring_buffer_write(uint8_t b)
{
  /*---------------------------------------------------------------
    Writes to a full buffer fail.
    -------------------------------------------------------------*/
  if(RING_BUFFER_FULL(rb)) { return 1; }

  /*---------------------------------------------------------------
    Room for at least one more byte.  Add it.
    -------------------------------------------------------------*/
  rb.buf[rb.wIdx] = b;
  rb.wIdx = (rb.wIdx + 1) % sizeof(rb.buf);
  return 0;
}

So now for the magic! Consider the case where we have two threads, a reader and a writer. We already guaranteed that accesses to the read and write positions will be atomic with the architecture rule. Since the reader doesn't touch the write position, and the writer doesn't touch the read position, there is no mutex required. All buffer accesses are free to happen as fast as the CPU can issue them.

The single reader, single writer case is exactly what happens in embedded world much of the time. For example, your serial port driver will have a transmit and receive buffer. Every write to the transmit buffer and read from the receive buffer will happen from the main loop. The serial port ISR will handle the opposing role for each buffer. Two threads of execution but still obeying the single reader, single writer case.

Even when scaled to larger systems we can still eliminate mutexes from the data structure. We just need to add an atomic increment operation to the requirements. It is slightly more complicated because a = (a + 1) % b translates into several machine instructions. You'll find the primitives required to make this work are often supported in the CPU architecture itself. Once again I defer to Wikipedia's phenomenal reference library. ARM CPUs have LDREX, STREX instructions. x86 CPUs always have to be difficult. Or you can use Windows' InterlockedIncrement() API.

So by making a few simplifying assumptions about functionality, we can guarantee thread safety in the single reader/writer case. Furthermore, by assuming a few more things about the CPU architecture or OS environment, we can guarantee thread safety in the multiple readers/writers case. Everything can be implemented without using mutual exclusion and winning on the performance that it saves.

Don't Quit

Slowly but surely, psychologists and parents are realizing that the most successful people in life got there because they were more persistent, and not smarter than everyone else. Especially for the gifted kids, as the New York article suggests, something they never teach you at school is the enormous value in determination and persistence.

For me at least, my elementary and high school careers were spent close to the top of my class. I was into math and science. I didn't have to work all that hard for amazing grades, and I don't remember putting in much effort. Sure I'd do my homework, and I was so proud of the hard fought 80% grade I earned in OAC English. But in subjects like Calculus, I started my homework from the back of the assignment where the problems were the most difficult, and worked until the questions became too easy.

Then came university, the big equalizer. All at once I was put in a class where everyone else was exactly the same as me; we were all the best and brightest from our towns, we were all very smart. So it quickly became those who had the biggest, strongest work ethic did the best. The course work was hard, even for us overachievers. Those who did all of their homework over the semester instead of veg'ing out to watch TV, did the best. When we graduated, they not only carried their amazing GPAs with them, but the habits of hard work and persistence that got them through five years of undergrad.

And so it continues to be seen in the workplace and everywhere else in life, whoever has the deepest well of determination reaps the rewards. Keep plugging away at something for days, for weeks, for months, and you'll eventually succeed. The brain is an incredible feat of engineering. Given enough time to train, it will automatically figure out the patterns and subtleties of success in anything that you want to achieve. Persistence is all that's required.

Fundamentally I guess the main reason for posting about this topic, is that I see it happen all the time at work. When confronted with a problem, they throw their hands up in resignation and look to a higher authority for direction. They give up as soon as the problem scope even hints that it's going to stray outside their area of expertise. It's so frustrating to think that all they need to do is take a little initiative and believe that things are still in their control. The real answer is there is always something more to be tried. The space you've explored so far will be forever finite. The possibilities that remain are wild and infinite, just waiting for someone persistent enough to tame them.

Saving Power with Protothreads

For those of you who haven't heard of it, Protothreads is a great library for adding co-operative multitasking to any C/C++ project. It allows you to easily separate complicated state machines into a set of loosely coupled thread routines. The library consists of just a few simple header files that are written in very portable C and C preprocessor code. Weighing in at only 2 bytes of overhead per thread, the library is especially suited for embedded systems where RAM and code space are valuable resources.

If you're already familiar with writing multithreaded code, then you'll easily adapt to the conventions introduced by the Protothreads library. This is a good thing. It means that anyone who needs to read your code (including you, six months down the road) will have less of a problem trying to figure out your intent. Normal threading concepts such as blocking waits and yielding are supported, as well as child threads and semaphores. It is up to you as designer to implement the scheduling algorithm, however it can be as simple as a round robin between all of the system threads.

When I was fine tuning a recent AVR project I wanted to lower current consumption by using the CPU's idle mode. Out of all the AVR's available sleep modes, idle mode is the easiest to use. It shuts down the CPU's clock but leaves all of the on chip devices active. This lets it wake up quickly from any interrupt but still save power when nothing interesting is happening. After a few tests I measured that it would reduce consumption by 2 mA from an active current consumption of 8 mA. An easy 25% savings.

The only issue is that the Protothreads library doesn't distinguish between threads that are yielding and threads that are blocked waiting. By convention, yielding is meant to give up processor time to lower priority threads. Once other tasks have had a chance to run, the yielding task expects to resume its work. This means we can't put the CPU to sleep. Waiting, on the other hand, is meant to block a thread until an event happens. Until the event occurs, the waiting thread doesn't need the CPU, so it's safe to sleep.

So I've made a small modification to the Protothreads library that allows the scheduler to figure out the state of each thread. Now scheduling and sleeping decisions are based on a more complete picture of the system's state. The return values for protothread thread switch operations have been changed to bit masks instead of simple enumerations. Now to find out if a thread is alive, you can use the provided PT_ALIVE_MASK and check that the result is non-zero.

#define PT_WAITING 1
#define PT_YIELDED 2
#define PT_EXITED  4
#define PT_ENDED   8

#define PT_ALIVE_MASK 3

PT_SCHEDULE() now returns the thread's state in a way that's compatible with the previous version. New code can check the value to see if a thread has exited, yielded or is waiting. Old code can continue to treat the result as a boolean value as before.

/**
 * Schedule a protothread.
 *
 * This function shedules a protothread. The return value of the
 * function is non-zero if the protothread is running or zero if the
 * protothread has exited.
 *
 * \param f The call to the C function implementing the protothread 
 * to be scheduled
 *
 * \hideinitializer
 */
#define PT_SCHEDULE(f) ((f) & PT_ALIVE_MASK)

If at least one thread has yielded, the scheduler must continue the main loop. Otherwise, if all threads are blocked waiting for an event, it's safe to put the CPU to sleep.

PT_WAIT_THREAD() now returns the child thread's state instead of defaulting to PT_WAITING. This will let the child thread resume processing once all other threads have had a chance to use the CPU. I had to unwrap the call to PT_WAIT_UNTIL() to access the thread's return value, but it should be invisible to existing code.

/**
 * Block and wait until a child protothread completes.
 *
 * This macro schedules a child protothread. The current protothread
 * will block until the child protothread completes.
 *
 * \note The child protothread must be manually initialized with the
 * PT_INIT() function before this function is used.
 *
 * \param pt A pointer to the protothread control structure.
 * \param thread The child protothread with arguments
 *
 * \sa PT_SPAWN()
 *
 * \hideinitializer
 */
#define PT_WAIT_THREAD(pt, thread)     \
  do {                                 \
    LC_SET((pt)->lc);                  \
    char c = PT_SCHEDULE(thread);      \
    if(c) {                            \
      return c;                        \
    }                                  \
  } while(0)

You can find the updated version of the Protothreads header file here.