Using a circular buffer is an effective way of handling concurrent-access problems: the best way to deal with concurrent access is to perform no concurrent access whatsoever.

The circular buffer uses an algorithm called ``producer and consumer''--one player pushes data in and the other pulls data out. Concurrent access is avoided if there is exactly one producer and exactly one consumer. There are two examples of producer and consumer in the short module. In one case, the reading process is waiting to consume data that is produced at interrupt time; in the other, the bottom half consumes data produced by the top half.

Two pointers are used to address a circular buffer: head and tail. head is the point at which data is being written and is updated only by the producer of the data. Data is being read from tail, which is updated only by the consumer. As I mentioned above, if data is written at interrupt time, you must be careful when accessing head multiple times. You should either declare it as volatile or disable interrupts before entering race conditions.

The circular buffer runs smoothly, except when it fills up. If that happens, things become hairy, and you can choose among different possible solutions. The short implementation just loses data; there’s no check for overflow, and if head goes beyond tail, a whole buffer of data is lost. Some alternative implementations are to drop the last item; overwrite the buffer tail, as printk does (see Section 4.1.2 in Chapter 4); hold up the producer, as scullpipe does; or allocate a temporary extra buffer to back up the main buffer. The best solution depends on the importance of your data and other situation-specific questions, so I won’t cover it here.

Although the circular buffer appears to solve the problem of concurrent access, there is still the possibility of a race condition when the read function goes to sleep. This code shows where the problem appears in short:

while (short_head == short_tail) {
    interruptible_sleep_on(&short_queue);
    /* ... */
}

When executing this statement, it is possible that new data will arrive after the while condition is evaluated as true and before the process goes to sleep. Information carried in by the interrupt won’t be read by the process; the process goes to sleep even though head != tail, and it isn’t awakened until the next data item arrives.

I didn’t implement correct locking for short because the source of short_read is included in Section 8.2.2 in Chapter 8, and at that point this discussion was not worth introducing. Also, the data involved is not worth the effort.

While the data that short collects is not vital, and the likelihood of getting an interrupt in the time lapse between two successive instructions is often negligible, sometimes you just can’t take the risk of going to sleep when data is pending.

This problem is general enough to deserve special treatment and is delayed to Section 9.7.4 later in this chapter, where I’ll discuss it in detail.

It’s interesting to note that only a producer and consumer situation can be addressed with a circular buffer. A programmer must often deal with more complex data structures to solve the concurrent-access problem. The producer/consumer situation is actually the simplest class of these problems; other structures, such as linked lists, simply don’t lend themselves to a circular buffer implementation.

A commonly used method of acquiring unique access to shared data is to call cli to disable interrupt reporting in the processor. Whenever a data item (like a linked list) is modified at interrupt time and by a function living in the normal computational flow, interrupts must be disabled by the latter function before it touches shared data.

The race condition in this case is between the instruction reading a shared item and the instruction using the knowledge just acquired. For example, the following loop may fail reading a linked list if the list is modified at interrupt time:

for (ptr = listHead; ptr; ptr = ptr->next)
    /* do something */ ;

An interrupt may change the value of ptr after it has been read but before it is used. If this happens, you’ll have problems as soon as you use ptr because the current value of the pointer is no longer related to the list.

One possible solution is to disable interrupts for the duration of the critical loop. Although the code for disabling interrupts was introduced in Chapter 2, in the section Section 2.5.2, it is worth repeating it here:

unsigned long flags;
save_flags(flags);
cli();
/* critical code */
restore_flags(flags);

As a matter of fact, in the device methods, the simpler cli/sti pair can be used instead because you can count on interrupts being enabled when a process enters a system call. However, you have to use the safer save_flags/restore_flags solution in a function that is called by other code, where you can’t make any assumptions about the current value of the interrupt flag.

The third approach to shared data variables is to use locks that are accessed through atomic instructions. Whenever one of two unrelated entities (such as an interrupt handler and the read system call, or two processors in a Symmetric Multi-Processor computer) need to access a shared data item, it must first acquire the lock. If the lock can’t be acquired, it must be waited for.

The Linux kernel exports two sets of functions to deal with locks: bit operations and access to the ``atomic'' data type.

/* try to set lock */
while (set_bit(nr, addr) != 0)
    wait_for_a_while();

/* do your work */

/* release lock, and check... */
if (clear_bit(nr, addr) == 0)
    something_went_wrong(); /* already released: error */

The one race condition that has been omitted so far in this discussion is the problem of going to sleep. It is a problem far more general than interrupt-driven I/O, and an efficient solution requires a little knowledge of the internals of sleep_on.

This particular race condition occurs between the time the condition is checked for going to sleep and the actual invocation of sleep_on. This test case is the same statement used above, but I feel it is worth showing once again:

while (short_head == short_tail) {
    interruptible_sleep_on(&short_queue);
    /* ... */
}

If the comparison and the going to sleep must be performed safely, first disable interrupt reporting and then test the condition and go to sleep. Thus, the variable being tested in the comparison cannot be modified. The kernel allows the process to go to sleep after issuing cli. The kernel simply reenables interrupt reporting just before calling schedule, after inserting the process into its wait queue.

The code examples introduced here use a while loop, which performs signal handling. If a blocked signal is reported to the process, interruptible_sleep_on returns, and the test in the while statement is performed again. The following is one possible implementation:

while (short_head == short_tail) {
    cli();
    if (short_head == short_tail)
        interruptible_sleep_on(&short_queue);
    sti();
    /* .... signal decoding .... */
}

If an interrupt happens after the cli, the interrupt remains pending while the current process is being put to sleep. By the time the interrupt is eventually reported to the processor, the process is already asleep and can be awakened safely.

In this case, I used cli/sti because the sample code is designed to live within the read method; the safer save_flags, cli, and restore_flags functions should be used otherwise.

If you don’t want to disable interrupt reporting while going to sleep, there is another way to perform the same job (one that Linus likes a lot). Anyway, you can skip the following discussion if you’d like, since it’s slightly elaborate.

The idea is that the process can add itself to the wait queue, declare itself to be sleeping, and then perform its tests. This is the typical implementation:

struct wait_queue wait = { current, NULL };

add_wait_queue(&short_queue, &wait);
do {
    current->state = TASK_INTERRUPTIBLE;
    schedule();
}
while ((short_head == short_tail)
        && !(current->signal & ~current->blocked) );
remove_wait_queue(&short_queue, &wait);

if (current->signal & ~current->blocked) /* a signal arrived */
    return -ERESTARTSYS;

This code is somewhat like an unrolling of the internals of sleep_on. The wait variable is explicitly declared because it is needed to put a process to sleep; this fact was explained in Section 5.2.3, in Chapter 5. This example introduced several new symbols:

With this background, let’s look at what happens when an interrupt arrives. The handler calls wake_up_interruptible(&short_queue); for Linux, this means ``set state to TASK_RUNNING.'' Therefore, if an interrupt is reported between the while condition and the call to schedule, the task will already have been marked as running, with no data loss.

If, on the other hand, the process is still ``interruptible,'' schedule leaves it sleeping.

It’s interesting to note that the wake_up call doesn’t remove the process from the wait queue. It’s sleep_on that adds and removes the process from the wait queue. The code must call add_wait_queue and remove_wait_queue explicitly, because sleep_on hasn’t been used in this case.