mirror of
https://github.com/LCTT/TranslateProject.git
synced 2025-01-25 23:11:02 +08:00
Merge pull request #13481 from FSSlc/master
[Translated] 20190416 Inter-process communication in Linux- Using pipes and message queues.md
This commit is contained in:
Xingyu.Wang
GitHub
commit
79d5e72fd5
@ -1,531 +0,0 @@
|
||||
[#]: collector: (lujun9972)
|
||||
[#]: translator: (FSSlc)
|
||||
[#]: reviewer: ( )
|
||||
[#]: publisher: ( )
|
||||
[#]: url: ( )
|
||||
[#]: subject: (Inter-process communication in Linux: Using pipes and message queues)
|
||||
[#]: via: (https://opensource.com/article/19/4/interprocess-communication-linux-channels)
|
||||
[#]: author: (Marty Kalin https://opensource.com/users/mkalindepauledu)
|
||||
|
||||
Inter-process communication in Linux: Using pipes and message queues
|
||||
======
|
||||
Learn how processes synchronize with each other in Linux.
|
||||
![Chat bubbles][1]
|
||||
|
||||
This is the second article in a series about [interprocess communication][2] (IPC) in Linux. The [first article][3] focused on IPC through shared storage: shared files and shared memory segments. This article turns to pipes, which are channels that connect processes for communication. A channel has a _write end_ for writing bytes, and a _read end_ for reading these bytes in FIFO (first in, first out) order. In typical use, one process writes to the channel, and a different process reads from this same channel. The bytes themselves might represent anything: numbers, employee records, digital movies, and so on.
|
||||
|
||||
Pipes come in two flavors, named and unnamed, and can be used either interactively from the command line or within programs; examples are forthcoming. This article also looks at memory queues, which have fallen out of fashion—but undeservedly so.
|
||||
|
||||
The code examples in the first article acknowledged the threat of race conditions (either file-based or memory-based) in IPC that uses shared storage. The question naturally arises about safe concurrency for the channel-based IPC, which will be covered in this article. The code examples for pipes and memory queues use APIs with the POSIX stamp of approval, and a core goal of the POSIX standards is thread-safety.
|
||||
|
||||
Consider the [man pages for the **mq_open**][4] function, which belongs to the memory queue API. These pages include a section on [Attributes][5] with this small table:
|
||||
|
||||
Interface | Attribute | Value
|
||||
---|---|---
|
||||
mq_open() | Thread safety | MT-Safe
|
||||
|
||||
The value **MT-Safe** (with **MT** for multi-threaded) means that the **mq_open** function is thread-safe, which in turn implies process-safe: A process executes in precisely the sense that one of its threads executes, and if a race condition cannot arise among threads in the _same_ process, such a condition cannot arise among threads in different processes. The **MT-Safe** attribute assures that a race condition does not arise in invocations of **mq_open**. In general, channel-based IPC is concurrent-safe, although a cautionary note is raised in the examples that follow.
|
||||
|
||||
### Unnamed pipes
|
||||
|
||||
Let's start with a contrived command line example that shows how unnamed pipes work. On all modern systems, the vertical bar **|** represents an unnamed pipe at the command line. Assume **%** is the command line prompt, and consider this command:
|
||||
|
||||
|
||||
```
|
||||
`% sleep 5 | echo "Hello, world!" ## writer to the left of |, reader to the right`
|
||||
```
|
||||
|
||||
The _sleep_ and _echo_ utilities execute as separate processes, and the unnamed pipe allows them to communicate. However, the example is contrived in that no communication occurs. The greeting _Hello, world!_ appears on the screen; then, after about five seconds, the command line prompt returns, indicating that both the _sleep_ and _echo_ processes have exited. What's going on?
|
||||
|
||||
In the vertical-bar syntax from the command line, the process to the left ( _sleep_ ) is the writer, and the process to the right ( _echo_ ) is the reader. By default, the reader blocks until there are bytes to read from the channel, and the writer—after writing its bytes—finishes up by sending an end-of-stream marker. (Even if the writer terminates prematurely, an end-of-stream marker is sent to the reader.) The unnamed pipe persists until both the writer and the reader terminate.
|
||||
|
||||
In the contrived example, the _sleep_ process does not write any bytes to the channel but does terminate after about five seconds, which sends an end-of-stream marker to the channel. In the meantime, the _echo_ process immediately writes the greeting to the standard output (the screen) because this process does not read any bytes from the channel, so it does no waiting. Once the _sleep_ and _echo_ processes terminate, the unnamed pipe—not used at all for communication—goes away and the command line prompt returns.
|
||||
|
||||
Here is a more useful example using two unnamed pipes. Suppose that the file _test.dat_ looks like this:
|
||||
|
||||
|
||||
```
|
||||
this
|
||||
is
|
||||
the
|
||||
way
|
||||
the
|
||||
world
|
||||
ends
|
||||
```
|
||||
|
||||
The command:
|
||||
|
||||
|
||||
```
|
||||
`% cat test.dat | sort | uniq`
|
||||
```
|
||||
|
||||
pipes the output from the _cat_ (concatenate) process into the _sort_ process to produce sorted output, and then pipes the sorted output into the _uniq_ process to eliminate duplicate records (in this case, the two occurrences of **the** reduce to one):
|
||||
|
||||
|
||||
```
|
||||
ends
|
||||
is
|
||||
the
|
||||
this
|
||||
way
|
||||
world
|
||||
```
|
||||
|
||||
The scene now is set for a program with two processes that communicate through an unnamed pipe.
|
||||
|
||||
#### Example 1. Two processes communicating through an unnamed pipe.
|
||||
|
||||
|
||||
```
|
||||
#include <sys/wait.h> /* wait */
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h> /* exit functions */
|
||||
#include <unistd.h> /* read, write, pipe, _exit */
|
||||
#include <string.h>
|
||||
|
||||
#define ReadEnd 0
|
||||
#define WriteEnd 1
|
||||
|
||||
void report_and_exit(const char* msg) {
|
||||
[perror][6](msg);
|
||||
[exit][7](-1); /** failure **/
|
||||
}
|
||||
|
||||
int main() {
|
||||
int pipeFDs[2]; /* two file descriptors */
|
||||
char buf; /* 1-byte buffer */
|
||||
const char* msg = "Nature's first green is gold\n"; /* bytes to write */
|
||||
|
||||
if (pipe(pipeFDs) < 0) report_and_exit("pipeFD");
|
||||
pid_t cpid = fork(); /* fork a child process */
|
||||
if (cpid < 0) report_and_exit("fork"); /* check for failure */
|
||||
|
||||
if (0 == cpid) { /*** child ***/ /* child process */
|
||||
close(pipeFDs[WriteEnd]); /* child reads, doesn't write */
|
||||
|
||||
while (read(pipeFDs[ReadEnd], &buf, 1) > 0) /* read until end of byte stream */
|
||||
write(STDOUT_FILENO, &buf, sizeof(buf)); /* echo to the standard output */
|
||||
|
||||
close(pipeFDs[ReadEnd]); /* close the ReadEnd: all done */
|
||||
_exit(0); /* exit and notify parent at once */
|
||||
}
|
||||
else { /*** parent ***/
|
||||
close(pipeFDs[ReadEnd]); /* parent writes, doesn't read */
|
||||
|
||||
write(pipeFDs[WriteEnd], msg, [strlen][8](msg)); /* write the bytes to the pipe */
|
||||
close(pipeFDs[WriteEnd]); /* done writing: generate eof */
|
||||
|
||||
wait(NULL); /* wait for child to exit */
|
||||
[exit][7](0); /* exit normally */
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
```
|
||||
|
||||
The _pipeUN_ program above uses the system function **fork** to create a process. Although the program has but a single source file, multi-processing occurs during (successful) execution. Here are the particulars in a quick review of how the library function **fork** works:
|
||||
|
||||
* The **fork** function, called in the _parent_ process, returns **-1** to the parent in case of failure. In the _pipeUN_ example, the call is: [code]`pid_t cpid = fork(); /* called in parent */`[/code] The returned value is stored, in this example, in the variable **cpid** of integer type **pid_t**. (Every process has its own _process ID_ , a non-negative integer that identifies the process.) Forking a new process could fail for several reasons, including a full _process table_ , a structure that the system maintains to track processes. Zombie processes, clarified shortly, can cause a process table to fill if these are not harvested.
|
||||
* If the **fork** call succeeds, it thereby spawns (creates) a new child process, returning one value to the parent but a different value to the child. Both the parent and the child process execute the _same_ code that follows the call to **fork**. (The child inherits copies of all the variables declared so far in the parent.) In particular, a successful call to **fork** returns:
|
||||
* Zero to the child process
|
||||
* The child's process ID to the parent
|
||||
* An _if/else_ or equivalent construct typically is used after a successful **fork** call to segregate code meant for the parent from code meant for the child. In this example, the construct is: [code] if (0 == cpid) { /*** child ***/
|
||||
...
|
||||
}
|
||||
else { /*** parent ***/
|
||||
...
|
||||
}
|
||||
```
|
||||
If forking a child succeeds, the _pipeUN_ program proceeds as follows. There is an integer array:
|
||||
```
|
||||
`int pipeFDs[2]; /* two file descriptors */`
|
||||
```
|
||||
to hold two file descriptors, one for writing to the pipe and another for reading from the pipe. (The array element **pipeFDs[0]** is the file descriptor for the read end, and the array element **pipeFDs[1]** is the file descriptor for the write end.) A successful call to the system **pipe** function, made immediately before the call to **fork** , populates the array with the two file descriptors:
|
||||
```
|
||||
`if (pipe(pipeFDs) < 0) report_and_exit("pipeFD");`
|
||||
```
|
||||
The parent and the child now have copies of both file descriptors, but the _separation of concerns_ pattern means that each process requires exactly one of the descriptors. In this example, the parent does the writing and the child does the reading, although the roles could be reversed. The first statement in the child _if_ -clause code, therefore, closes the pipe's write end:
|
||||
```
|
||||
`close(pipeFDs[WriteEnd]); /* called in child code */`
|
||||
```
|
||||
and the first statement in the parent _else_ -clause code closes the pipe's read end:
|
||||
```
|
||||
`close(pipeFDs[ReadEnd]); /* called in parent code */`
|
||||
```
|
||||
The parent then writes some bytes (ASCII codes) to the unnamed pipe, and the child reads these and echoes them to the standard output.
|
||||
|
||||
One more aspect of the program needs clarification: the call to the **wait** function in the parent code. Once spawned, a child process is largely independent of its parent, as even the short _pipeUN_ program illustrates. The child can execute arbitrary code that may have nothing to do with the parent. However, the system does notify the parent through a signal—if and when the child terminates.
|
||||
|
||||
What if the parent terminates before the child? In this case, unless precautions are taken, the child becomes and remains a _zombie_ process with an entry in the process table. The precautions are of two broad types. One precaution is to have the parent notify the system that the parent has no interest in the child's termination:
|
||||
```
|
||||
`signal(SIGCHLD, SIG_IGN); /* in parent: ignore notification */`
|
||||
```
|
||||
A second approach is to have the parent execute a **wait** on the child's termination, thereby ensuring that the parent outlives the child. This second approach is used in the _pipeUN_ program, where the parent code has this call:
|
||||
```
|
||||
`wait(NULL); /* called in parent */`
|
||||
```
|
||||
This call to **wait** means _wait until the termination of any child occurs_ , and in the _pipeUN_ program, there is only one child process. (The **NULL** argument could be replaced with the address of an integer variable to hold the child's exit status.) There is a more flexible **waitpid** function for fine-grained control, e.g., for specifying a particular child process among several.
|
||||
|
||||
The _pipeUN_ program takes another precaution. When the parent is done waiting, the parent terminates with the call to the regular **exit** function. By contrast, the child terminates with a call to the **_exit** variant, which fast-tracks notification of termination. In effect, the child is telling the system to notify the parent ASAP that the child has terminated.
|
||||
|
||||
If two processes write to the same unnamed pipe, can the bytes be interleaved? For example, if process P1 writes:
|
||||
```
|
||||
`foo bar`
|
||||
```
|
||||
to a pipe and process P2 concurrently writes:
|
||||
```
|
||||
`baz baz`
|
||||
```
|
||||
to the same pipe, it seems that the pipe contents might be something arbitrary, such as:
|
||||
```
|
||||
`baz foo baz bar`
|
||||
```
|
||||
The POSIX standard ensures that writes are not interleaved so long as no write exceeds **PIPE_BUF** bytes. On Linux systems, **PIPE_BUF** is 4,096 bytes in size. My preference with pipes is to have a single writer and a single reader, thereby sidestepping the issue.
|
||||
|
||||
## Named pipes
|
||||
|
||||
An unnamed pipe has no backing file: the system maintains an in-memory buffer to transfer bytes from the writer to the reader. Once the writer and reader terminate, the buffer is reclaimed, so the unnamed pipe goes away. By contrast, a named pipe has a backing file and a distinct API.
|
||||
|
||||
Let's look at another command line example to get the gist of named pipes. Here are the steps:
|
||||
|
||||
* Open two terminals. The working directory should be the same for both.
|
||||
* In one of the terminals, enter these two commands (the prompt again is **%** , and my comments start with **##** ): [code] % mkfifo tester ## creates a backing file named tester
|
||||
% cat tester ## type the pipe's contents to stdout [/code] At the beginning, nothing should appear in the terminal because nothing has been written yet to the named pipe.
|
||||
* In the second terminal, enter the command: [code] % cat > tester ## redirect keyboard input to the pipe
|
||||
hello, world! ## then hit Return key
|
||||
bye, bye ## ditto
|
||||
<Control-C> ## terminate session with a Control-C [/code] Whatever is typed into this terminal is echoed in the other. Once **Ctrl+C** is entered, the regular command line prompt returns in both terminals: the pipe has been closed.
|
||||
* Clean up by removing the file that implements the named pipe: [code]`% unlink tester`
|
||||
```
|
||||
|
||||
|
||||
|
||||
As the utility's name _mkfifo_ implies, a named pipe also is called a FIFO because the first byte in is the first byte out, and so on. There is a library function named **mkfifo** that creates a named pipe in programs and is used in the next example, which consists of two processes: one writes to the named pipe and the other reads from this pipe.
|
||||
|
||||
#### Example 2. The _fifoWriter_ program
|
||||
|
||||
|
||||
```
|
||||
#include <sys/types.h>
|
||||
#include <sys/stat.h>
|
||||
#include <fcntl.h>
|
||||
#include <unistd.h>
|
||||
#include <time.h>
|
||||
#include <stdlib.h>
|
||||
#include <stdio.h>
|
||||
|
||||
#define MaxLoops 12000 /* outer loop */
|
||||
#define ChunkSize 16 /* how many written at a time */
|
||||
#define IntsPerChunk 4 /* four 4-byte ints per chunk */
|
||||
#define MaxZs 250 /* max microseconds to sleep */
|
||||
|
||||
int main() {
|
||||
const char* pipeName = "./fifoChannel";
|
||||
mkfifo(pipeName, 0666); /* read/write for user/group/others */
|
||||
int fd = open(pipeName, O_CREAT | O_WRONLY); /* open as write-only */
|
||||
if (fd < 0) return -1; /* can't go on */
|
||||
|
||||
int i;
|
||||
for (i = 0; i < MaxLoops; i++) { /* write MaxWrites times */
|
||||
int j;
|
||||
for (j = 0; j < ChunkSize; j++) { /* each time, write ChunkSize bytes */
|
||||
int k;
|
||||
int chunk[IntsPerChunk];
|
||||
for (k = 0; k < IntsPerChunk; k++)
|
||||
chunk[k] = [rand][9]();
|
||||
write(fd, chunk, sizeof(chunk));
|
||||
}
|
||||
usleep(([rand][9]() % MaxZs) + 1); /* pause a bit for realism */
|
||||
}
|
||||
|
||||
close(fd); /* close pipe: generates an end-of-stream marker */
|
||||
unlink(pipeName); /* unlink from the implementing file */
|
||||
[printf][10]("%i ints sent to the pipe.\n", MaxLoops * ChunkSize * IntsPerChunk);
|
||||
|
||||
return 0;
|
||||
}
|
||||
```
|
||||
|
||||
The _fifoWriter_ program above can be summarized as follows:
|
||||
|
||||
* The program creates a named pipe for writing: [code] mkfifo(pipeName, 0666); /* read/write perms for user/group/others */
|
||||
int fd = open(pipeName, O_CREAT | O_WRONLY); [/code] where **pipeName** is the name of the backing file passed to **mkfifo** as the first argument. The named pipe then is opened with the by-now familiar call to the **open** function, which returns a file descriptor.
|
||||
* For a touch of realism, the _fifoWriter_ does not write all the data at once, but instead writes a chunk, sleeps a random number of microseconds, and so on. In total, 768,000 4-byte integer values are written to the named pipe.
|
||||
* After closing the named pipe, the _fifoWriter_ also unlinks the file: [code] close(fd); /* close pipe: generates end-of-stream marker */
|
||||
unlink(pipeName); /* unlink from the implementing file */ [/code] The system reclaims the backing file once every process connected to the pipe has performed the unlink operation. In this example, there are only two such processes: the _fifoWriter_ and the _fifoReader_ , both of which do an _unlink_ operation.
|
||||
|
||||
|
||||
|
||||
The two programs should be executed in different terminals with the same working directory. However, the _fifoWriter_ should be started before the _fifoReader_ , as the former creates the pipe. The _fifoReader_ then accesses the already created named pipe.
|
||||
|
||||
#### Example 3. The _fifoReader_ program
|
||||
|
||||
|
||||
```
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <fcntl.h>
|
||||
#include <unistd.h>
|
||||
|
||||
unsigned is_prime(unsigned n) { /* not pretty, but efficient */
|
||||
if (n <= 3) return n > 1;
|
||||
if (0 == (n % 2) || 0 == (n % 3)) return 0;
|
||||
|
||||
unsigned i;
|
||||
for (i = 5; (i * i) <= n; i += 6)
|
||||
if (0 == (n % i) || 0 == (n % (i + 2))) return 0;
|
||||
|
||||
return 1; /* found a prime! */
|
||||
}
|
||||
|
||||
int main() {
|
||||
const char* file = "./fifoChannel";
|
||||
int fd = open(file, O_RDONLY);
|
||||
if (fd < 0) return -1; /* no point in continuing */
|
||||
unsigned count = 0, total = 0, primes_count = 0;
|
||||
|
||||
while (1) {
|
||||
int next;
|
||||
int i;
|
||||
|
||||
ssize_t count = read(fd, &next, sizeof(int));
|
||||
if (0 == count) break; /* end of stream */
|
||||
else if (count == sizeof(int)) { /* read a 4-byte int value */
|
||||
total++;
|
||||
if (is_prime(next)) primes_count++;
|
||||
}
|
||||
}
|
||||
|
||||
close(fd); /* close pipe from read end */
|
||||
unlink(file); /* unlink from the underlying file */
|
||||
[printf][10]("Received ints: %u, primes: %u\n", total, primes_count);
|
||||
|
||||
return 0;
|
||||
}
|
||||
```
|
||||
|
||||
The _fifoReader_ program above can be summarized as follows:
|
||||
|
||||
* Because the _fifoWriter_ creates the named pipe, the _fifoReader_ needs only the standard call **open** to access the pipe through the backing file: [code] const char* file = "./fifoChannel";
|
||||
int fd = open(file, O_RDONLY); [/code] The file opens as read-only.
|
||||
* The program then goes into a potentially infinite loop, trying to read a 4-byte chunk on each iteration. The **read** call: [code]`ssize_t count = read(fd, &next, sizeof(int));`[/code] returns 0 to indicate end-of-stream, in which case the _fifoReader_ breaks out of the loop, closes the named pipe, and unlinks the backing file before terminating.
|
||||
* After reading a 4-byte integer, the _fifoReader_ checks whether the number is a prime. This represents the business logic that a production-grade reader might perform on the received bytes. On a sample run, there were 37,682 primes among the 768,000 integers received.
|
||||
|
||||
|
||||
|
||||
On repeated sample runs, the _fifoReader_ successfully read all of the bytes that the _fifoWriter_ wrote. This is not surprising. The two processes execute on the same host, taking network issues out of the equation. Named pipes are a highly reliable and efficient IPC mechanism and, therefore, in wide use.
|
||||
|
||||
Here is the output from the two programs, each launched from a separate terminal but with the same working directory:
|
||||
|
||||
|
||||
```
|
||||
% ./fifoWriter
|
||||
768000 ints sent to the pipe.
|
||||
###
|
||||
% ./fifoReader
|
||||
Received ints: 768000, primes: 37682
|
||||
```
|
||||
|
||||
### Message queues
|
||||
|
||||
Pipes have strict FIFO behavior: the first byte written is the first byte read, the second byte written is the second byte read, and so forth. Message queues can behave in the same way but are flexible enough that byte chunks can be retrieved out of FIFO order.
|
||||
|
||||
As the name suggests, a message queue is a sequence of messages, each of which has two parts:
|
||||
|
||||
* The payload, which is an array of bytes ( **char** in C)
|
||||
* A type, given as a positive integer value; types categorize messages for flexible retrieval
|
||||
|
||||
|
||||
|
||||
Consider the following depiction of a message queue, with each message labeled with an integer type:
|
||||
|
||||
|
||||
```
|
||||
+-+ +-+ +-+ +-+
|
||||
sender--->|3|--->|2|--->|2|--->|1|--->receiver
|
||||
+-+ +-+ +-+ +-+
|
||||
```
|
||||
|
||||
Of the four messages shown, the one labeled 1 is at the front, i.e., closest to the receiver. Next come two messages with label 2, and finally, a message labeled 3 at the back. If strict FIFO behavior were in play, then the messages would be received in the order 1-2-2-3. However, the message queue allows other retrieval orders. For example, the messages could be retrieved by the receiver in the order 3-2-1-2.
|
||||
|
||||
The _mqueue_ example consists of two programs, the _sender_ that writes to the message queue and the _receiver_ that reads from this queue. Both programs include the header file _queue.h_ shown below:
|
||||
|
||||
#### Example 4. The header file _queue.h_
|
||||
|
||||
|
||||
```
|
||||
#define ProjectId 123
|
||||
#define PathName "queue.h" /* any existing, accessible file would do */
|
||||
#define MsgLen 4
|
||||
#define MsgCount 6
|
||||
|
||||
typedef struct {
|
||||
long type; /* must be of type long */
|
||||
char payload[MsgLen + 1]; /* bytes in the message */
|
||||
} queuedMessage;
|
||||
```
|
||||
|
||||
The header file defines a structure type named **queuedMessage** , with **payload** (byte array) and **type** (integer) fields. This file also defines symbolic constants (the **#define** statements), the first two of which are used to generate a key that, in turn, is used to get a message queue ID. The **ProjectId** can be any positive integer value, and the **PathName** must be an existing, accessible file—in this case, the file _queue.h_. The setup statements in both the _sender_ and the _receiver_ programs are:
|
||||
|
||||
|
||||
```
|
||||
key_t key = ftok(PathName, ProjectId); /* generate key */
|
||||
int qid = msgget(key, 0666 | IPC_CREAT); /* use key to get queue id */
|
||||
```
|
||||
|
||||
The ID **qid** is, in effect, the counterpart of a file descriptor for message queues.
|
||||
|
||||
#### Example 5. The message _sender_ program
|
||||
|
||||
|
||||
```
|
||||
#include <stdio.h>
|
||||
#include <sys/ipc.h>
|
||||
#include <sys/msg.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include "queue.h"
|
||||
|
||||
void report_and_exit(const char* msg) {
|
||||
[perror][6](msg);
|
||||
[exit][7](-1); /* EXIT_FAILURE */
|
||||
}
|
||||
|
||||
int main() {
|
||||
key_t key = ftok(PathName, ProjectId);
|
||||
if (key < 0) report_and_exit("couldn't get key...");
|
||||
|
||||
int qid = msgget(key, 0666 | IPC_CREAT);
|
||||
if (qid < 0) report_and_exit("couldn't get queue id...");
|
||||
|
||||
char* payloads[] = {"msg1", "msg2", "msg3", "msg4", "msg5", "msg6"};
|
||||
int types[] = {1, 1, 2, 2, 3, 3}; /* each must be > 0 */
|
||||
int i;
|
||||
for (i = 0; i < MsgCount; i++) {
|
||||
/* build the message */
|
||||
queuedMessage msg;
|
||||
msg.type = types[i];
|
||||
[strcpy][11](msg.payload, payloads[i]);
|
||||
|
||||
/* send the message */
|
||||
msgsnd(qid, &msg, sizeof(msg), IPC_NOWAIT); /* don't block */
|
||||
[printf][10]("%s sent as type %i\n", msg.payload, (int) msg.type);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
```
|
||||
|
||||
The _sender_ program above sends out six messages, two each of a specified type: the first messages are of type 1, the next two of type 2, and the last two of type 3. The sending statement:
|
||||
|
||||
|
||||
```
|
||||
`msgsnd(qid, &msg, sizeof(msg), IPC_NOWAIT);`
|
||||
```
|
||||
|
||||
is configured to be non-blocking (the flag **IPC_NOWAIT** ) because the messages are so small. The only danger is that a full queue, unlikely in this example, would result in a sending failure. The _receiver_ program below also receives messages using the **IPC_NOWAIT** flag.
|
||||
|
||||
#### Example 6. The message _receiver_ program
|
||||
|
||||
|
||||
```
|
||||
#include <stdio.h>
|
||||
#include <sys/ipc.h>
|
||||
#include <sys/msg.h>
|
||||
#include <stdlib.h>
|
||||
#include "queue.h"
|
||||
|
||||
void report_and_exit(const char* msg) {
|
||||
[perror][6](msg);
|
||||
[exit][7](-1); /* EXIT_FAILURE */
|
||||
}
|
||||
|
||||
int main() {
|
||||
key_t key= ftok(PathName, ProjectId); /* key to identify the queue */
|
||||
if (key < 0) report_and_exit("key not gotten...");
|
||||
|
||||
int qid = msgget(key, 0666 | IPC_CREAT); /* access if created already */
|
||||
if (qid < 0) report_and_exit("no access to queue...");
|
||||
|
||||
int types[] = {3, 1, 2, 1, 3, 2}; /* different than in sender */
|
||||
int i;
|
||||
for (i = 0; i < MsgCount; i++) {
|
||||
queuedMessage msg; /* defined in queue.h */
|
||||
if (msgrcv(qid, &msg, sizeof(msg), types[i], MSG_NOERROR | IPC_NOWAIT) < 0)
|
||||
[puts][12]("msgrcv trouble...");
|
||||
[printf][10]("%s received as type %i\n", msg.payload, (int) msg.type);
|
||||
}
|
||||
|
||||
/** remove the queue **/
|
||||
if (msgctl(qid, IPC_RMID, NULL) < 0) /* NULL = 'no flags' */
|
||||
report_and_exit("trouble removing queue...");
|
||||
|
||||
return 0;
|
||||
}
|
||||
```
|
||||
|
||||
The _receiver_ program does not create the message queue, although the API suggests as much. In the _receiver_ , the call:
|
||||
|
||||
|
||||
```
|
||||
`int qid = msgget(key, 0666 | IPC_CREAT);`
|
||||
```
|
||||
|
||||
is misleading because of the **IPC_CREAT** flag, but this flag really means _create if needed, otherwise access_. The _sender_ program calls **msgsnd** to send messages, whereas the _receiver_ calls **msgrcv** to retrieve them. In this example, the _sender_ sends the messages in the order 1-1-2-2-3-3, but the _receiver_ then retrieves them in the order 3-1-2-1-3-2, showing that message queues are not bound to strict FIFO behavior:
|
||||
|
||||
|
||||
```
|
||||
% ./sender
|
||||
msg1 sent as type 1
|
||||
msg2 sent as type 1
|
||||
msg3 sent as type 2
|
||||
msg4 sent as type 2
|
||||
msg5 sent as type 3
|
||||
msg6 sent as type 3
|
||||
|
||||
% ./receiver
|
||||
msg5 received as type 3
|
||||
msg1 received as type 1
|
||||
msg3 received as type 2
|
||||
msg2 received as type 1
|
||||
msg6 received as type 3
|
||||
msg4 received as type 2
|
||||
```
|
||||
|
||||
The output above shows that the _sender_ and the _receiver_ can be launched from the same terminal. The output also shows that the message queue persists even after the _sender_ process creates the queue, writes to it, and exits. The queue goes away only after the _receiver_ process explicitly removes it with the call to **msgctl** :
|
||||
|
||||
|
||||
```
|
||||
`if (msgctl(qid, IPC_RMID, NULL) < 0) /* remove queue */`
|
||||
```
|
||||
|
||||
### Wrapping up
|
||||
|
||||
The pipes and message queue APIs are fundamentally _unidirectional_ : one process writes and another reads. There are implementations of bidirectional named pipes, but my two cents is that this IPC mechanism is at its best when it is simplest. As noted earlier, message queues have fallen in popularity—but without good reason; these queues are yet another tool in the IPC toolbox. Part 3 completes this quick tour of the IPC toolbox with code examples of IPC through sockets and signals.
|
||||
|
||||
--------------------------------------------------------------------------------
|
||||
|
||||
via: https://opensource.com/article/19/4/interprocess-communication-linux-channels
|
||||
|
||||
作者：[Marty Kalin][a]
|
||||
选题：[lujun9972][b]
|
||||
译者：[译者ID](https://github.com/译者ID)
|
||||
校对：[校对者ID](https://github.com/校对者ID)
|
||||
|
||||
本文由 [LCTT](https://github.com/LCTT/TranslateProject) 原创编译，[Linux中国](https://linux.cn/) 荣誉推出
|
||||
|
||||
[a]: https://opensource.com/users/mkalindepauledu
|
||||
[b]: https://github.com/lujun9972
|
||||
[1]: https://opensource.com/sites/default/files/styles/image-full-size/public/lead-images/talk_chat_communication_team.png?itok=CYfZ_gE7 (Chat bubbles)
|
||||
[2]: https://en.wikipedia.org/wiki/Inter-process_communication
|
||||
[3]: https://opensource.com/article/19/4/interprocess-communication-ipc-linux-part-1
|
||||
[4]: http://man7.org/linux/man-pages/man2/mq_open.2.html
|
||||
[5]: http://man7.org/linux/man-pages/man2/mq_open.2.html#ATTRIBUTES
|
||||
[6]: http://www.opengroup.org/onlinepubs/009695399/functions/perror.html
|
||||
[7]: http://www.opengroup.org/onlinepubs/009695399/functions/exit.html
|
||||
[8]: http://www.opengroup.org/onlinepubs/009695399/functions/strlen.html
|
||||
[9]: http://www.opengroup.org/onlinepubs/009695399/functions/rand.html
|
||||
[10]: http://www.opengroup.org/onlinepubs/009695399/functions/printf.html
|
||||
[11]: http://www.opengroup.org/onlinepubs/009695399/functions/strcpy.html
|
||||
[12]: http://www.opengroup.org/onlinepubs/009695399/functions/puts.html
|
||||
@ -0,0 +1,574 @@
|
||||
[#]: collector: (lujun9972)
|
||||
[#]: translator: (FSSlc)
|
||||
[#]: reviewer: ( )
|
||||
[#]: publisher: ( )
|
||||
[#]: url: ( )
|
||||
[#]: subject: (Inter-process communication in Linux: Using pipes and message queues)
|
||||
[#]: via: (https://opensource.com/article/19/4/interprocess-communication-linux-channels)
|
||||
[#]: author: (Marty Kalin https://opensource.com/users/mkalindepauledu)
|
||||
|
||||
Linux 下的进程间通信：使用管道和消息队列
|
||||
======
|
||||
学习在 Linux 中进程是如何与其他进程进行同步的。
|
||||
![Chat bubbles][1]
|
||||
|
||||
本篇是 Linux 下[进程间通信][2]（IPC）系列的第二篇文章。[第一篇文章][3] 聚焦于通过共享文件和共享内存段这样的共享存储来进行 IPC。这篇文件的重点将转向管道，它是连接需要通信的进程之间的通道。管道拥有一个 _写端_ 用于写入字节数据，还有一个 _读端_ 用于按照先入先出的顺序读入这些字节数据。而这些字节数据可能代表任何东西：数字、数字电影等等。
|
||||
|
||||
管道有两种类型，命名管道和无名管道，都可以交互式的在命令行或程序中使用它们；相关的例子在下面展示。这篇文章也将介绍内存队列，尽管它们有些过时了，但它们不应该受这样的待遇。
|
||||
|
||||
在本系列的第一篇文章中的示例代码承认了在 IPC 中可能受到竞争条件（不管是基于文件的还是基于内存的）的威胁。自然地我们也会考虑基于管道的 IPC 的安全并发问题，这个也将在本文中提及。针对管道和内存队列的例子将会使用 POSIX 推荐使用的 API， POSIX 的一个核心目标就是线程安全。
|
||||
|
||||
考虑查看 [**mq_open** 函数的 man 页][4]，这个函数属于内存队列的 API。这个 man 页中有一个章节是有关 [特性][5] 的小表格：
|
||||
|
||||
Interface | Attribute | Value
|
||||
---|---|---
|
||||
mq_open() | Thread safety | MT-Safe
|
||||
|
||||
上面的 **MT-Safe**（**MT** 指的是 multi-threaded，多线程）意味着 **mq_open** 函数是线程安全的，进而暗示是进程安全的：一个进程的执行和它的一个线程执行的过程类似，假如竞争条件不会发生在处于 _相同_ 进程的线程中，那么这样的条件也不会发生在处于不同进程的线程中。**MT-Safe** 特性保证了调用 **mq_open** 时不会出现竞争条件。一般来说，基于通道的 IPC 是并发安全的，尽管在下面例子中会出现一个有关警告的注意事项。
|
||||
|
||||
### 无名管道
|
||||
|
||||
首先让我们通过一个特意构造的命令行例子来展示无名管道是如何工作的。在所有的现代系统中，符号 **|** 在命令行中都代表一个无名管道。假设我们的命令行提示符为 **%**，接下来考虑下面的命令：
|
||||
|
||||
```shell
|
||||
% sleep 5 | echo "Hello, world!" ## writer to the left of |, reader to the right
|
||||
```
|
||||
|
||||
_sleep_ 和 _echo_ 程序以不同的进程执行，无名管道允许它们进行通信。但是上面的例子被特意设计为没有通信发生。问候语 _Hello, world!_ 出现在屏幕中，然后过了 5 秒后，命令行返回，暗示 _sleep_ 和 _echo_ 进程都已经结束了。这期间发生了什么呢？
|
||||
|
||||
在命令行中的竖线 **|** 的语法中，左边的进程（_sleep_）是写入方，右边的进程（_echo_）为读取方。默认情况下，读入方将会堵塞，直到从通道中能够读取字节数据，而写入方在写完它的字节数据后，将发送 流已终止 的标志。（即便写入方永久地停止了，一个流已终止的标志还是会发给读取方。）无名管道将保持到写入方和读取方都停止的那个时刻。
|
||||
|
||||
在上面的例子中，_sleep_ 进程并没有向通道写入任何的字节数据，但在 5 秒后就停止了，这时将向通道发送一个流已终止的标志。与此同时，_echo_ 进程立即向标准输出（屏幕）写入问候语，因为这个进程并不从通道中读入任何字节，所以它并没有等待。一旦 _sleep_ 和 _echo_ 进程都终止了，不会再用作通信的无名管道将会消失然后返回命令行提示符。
|
||||
|
||||
下面这个更加实用示例将使用两个无名管道。我们假定文件 _test.dat_ 的内容如下：
|
||||
|
||||
```
|
||||
this
|
||||
is
|
||||
the
|
||||
way
|
||||
the
|
||||
world
|
||||
ends
|
||||
```
|
||||
|
||||
下面的命令
|
||||
|
||||
```
|
||||
% cat test.dat | sort | uniq
|
||||
```
|
||||
|
||||
会将 _cat_（concatenate 的缩写）进程的输出通过管道传给 _sort_ 进程以生成排序后的输出，然后将排序后的输出通过管道传给 _uniq_ 进程以消除重复的记录（在本例中，会将两次出现的 **the** 缩减为一个）：
|
||||
|
||||
```
|
||||
ends
|
||||
is
|
||||
the
|
||||
this
|
||||
way
|
||||
world
|
||||
```
|
||||
|
||||
下面展示的情景展示的是一个带有两个进程的程序通过一个无名管道通信来进行通信。
|
||||
|
||||
#### 示例 1. 两个进程通过一个无名管道来进行通信
|
||||
|
||||
|
||||
```c
|
||||
#include <sys/wait.h> /* wait */
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h> /* exit functions */
|
||||
#include <unistd.h> /* read, write, pipe, _exit */
|
||||
#include <string.h>
|
||||
|
||||
#define ReadEnd 0
|
||||
#define WriteEnd 1
|
||||
|
||||
void report_and_exit(const char* msg) {
|
||||
[perror][6](msg);
|
||||
[exit][7](-1); /** failure **/
|
||||
}
|
||||
|
||||
int main() {
|
||||
int pipeFDs[2]; /* two file descriptors */
|
||||
char buf; /* 1-byte buffer */
|
||||
const char* msg = "Nature's first green is gold\n"; /* bytes to write */
|
||||
|
||||
if (pipe(pipeFDs) < 0) report_and_exit("pipeFD");
|
||||
pid_t cpid = fork(); /* fork a child process */
|
||||
if (cpid < 0) report_and_exit("fork"); /* check for failure */
|
||||
|
||||
if (0 == cpid) { /*** child ***/ /* child process */
|
||||
close(pipeFDs[WriteEnd]); /* child reads, doesn't write */
|
||||
|
||||
while (read(pipeFDs[ReadEnd], &buf, 1) > 0) /* read until end of byte stream */
|
||||
write(STDOUT_FILENO, &buf, sizeof(buf)); /* echo to the standard output */
|
||||
|
||||
close(pipeFDs[ReadEnd]); /* close the ReadEnd: all done */
|
||||
_exit(0); /* exit and notify parent at once */
|
||||
}
|
||||
else { /*** parent ***/
|
||||
close(pipeFDs[ReadEnd]); /* parent writes, doesn't read */
|
||||
|
||||
write(pipeFDs[WriteEnd], msg, [strlen][8](msg)); /* write the bytes to the pipe */
|
||||
close(pipeFDs[WriteEnd]); /* done writing: generate eof */
|
||||
|
||||
wait(NULL); /* wait for child to exit */
|
||||
[exit][7](0); /* exit normally */
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
```
|
||||
|
||||
上面名为 _pipeUN_ 的程序使用系统函数 **fork** 来创建一个进程。尽管这个程序只有一个单一的源文件，在它正确执行的情况下将会发生多进程的情况。下面的内容是对库函数 **fork** 如何工作的一个简要回顾：
|
||||
|
||||
* **fork** 函数由 _父_ 进程调用，在失败时返回 **-1** 给父进程。在 _pipeUN_ 这个例子中，相应的调用是
|
||||
|
||||
```c
|
||||
pid_t cpid = fork(); /* called in parent */
|
||||
```
|
||||
|
||||
函数，调用后的返回值也被保存下来了。在这个例子中，保存在整数类型 **pid_t** 的变量 **cpid** 中。（每个进程有它自己的 _进程 ID_，一个非负的整数，用来标记进程）。复刻一个新的进程可能会因为多种原因而失败。最终它们将会被包括进一个完整的 _进程表_，这个结构由系统维持，以此来追踪进程状态。明确地说，僵尸进程假如没有被处理掉，将可能引起一个进程表被填满。
|
||||
* 假如 **fork** 调用成功，则它将创建一个新的子进程，向父进程返回一个值，向子进程返回另外的一个值。在调用 **fork** 后父进程和子进程都将执行相同的代码。（子进程继承了到此为止父进程中声明的所有变量的拷贝），特别地，一次成功的 **fork** 调用将返回如下的东西：
|
||||
* 向子进程返回 0
|
||||
* 向父进程返回子进程的进程 ID
|
||||
* 在依次成功的 **fork** 调用后，一个 _if/else_ 或等价的结构将会被用来隔离针对父进程和子进程的代码。在这个例子中，相应的声明为：
|
||||
|
||||
```c
|
||||
if (0 == cpid) { /*** child ***/
|
||||
...
|
||||
}
|
||||
else { /*** parent ***/
|
||||
...
|
||||
}
|
||||
```
|
||||
|
||||
假如成功地复刻出了一个子进程，_pipeUN_ 程序将像下面这样去执行。存在一个整数的数列
|
||||
|
||||
```c
|
||||
int pipeFDs[2]; /* two file descriptors */
|
||||
```
|
||||
|
||||
来保存两个文件描述符，一个用来向管道中写入，另一个从管道中写入。（数组元素 **pipeFDs[0]** 是读端的文件描述符，元素 **pipeFDs[1]** 是写端的文件描述符。）在调用 **fork** 之前，对系统 **pipe** 函数的成功调用，将立刻使得这个数组获得两个文件描述符：
|
||||
|
||||
```c
|
||||
if (pipe(pipeFDs) < 0) report_and_exit("pipeFD");
|
||||
```
|
||||
|
||||
父进程和子进程现在都有了文件描述符的副本。但 _分离关注点_ 模式意味着每个进程恰好只需要一个描述符。在这个例子中，父进程负责写入，而子进程负责读取，尽管这样的角色分配可以反过来。在 _if_ 子句中的第一个语句将用于关闭管道的读端：
|
||||
|
||||
```c
|
||||
close(pipeFDs[WriteEnd]); /* called in child code */
|
||||
```
|
||||
|
||||
在父进程中的 _else_ 子句将会关闭管道的读端：
|
||||
|
||||
```c
|
||||
close(pipeFDs[ReadEnd]); /* called in parent code */
|
||||
```
|
||||
|
||||
然后父进程将向无名管道中写入某些字节数据（ASCII 代码），子进程读取这些数据，然后向标准输出中回放它们。
|
||||
|
||||
在这个程序中还需要澄清的一点是在父进程代码中的 **wait** 函数。一旦被创建后，子进程很大程度上独立于它的父进程，正如简短的 _pipeUN_ 程序所展示的那样。子进程可以执行任意的代码，而它们可能与父进程完全没有关系。但是，假如当子进程终止时，系统将会通过一个信号来通知父进程。
|
||||
|
||||
要是父进程在子进程之前终止又该如何呢？在这种情形下，除非采取了预防措施，子进程将会变成在进程表中的一个 _僵尸_ 进程。预防措施有两大类型：第一种是让父进程去通知系统，告诉系统它对子进程的终止没有任何兴趣：
|
||||
|
||||
```c
|
||||
signal(SIGCHLD, SIG_IGN); /* in parent: ignore notification */
|
||||
```
|
||||
|
||||
第二种方法是在子进程终止时，让父进程执行一个 **wait**。这样就确保了父进程可以独立于子进程而存在。在 _pipeUN_ 程序中使用了第二种方法，其中父进程的代码使用的是下面的调用：
|
||||
|
||||
```c
|
||||
wait(NULL); /* called in parent */
|
||||
```
|
||||
|
||||
这个对 **wait** 的调用意味着 _一直等待直到任意一个子进程的终止发生_，因此在 _pipeUN_ 程序中，只有一个子进程。（其中的 **NULL** 参数可以被替换为一个保存有子程序退出状态的整数变量的地址。）对于更细颗粒度的控制，还可以使用更灵活的 **waitpid** 函数，例如特别指定多个子进程中的某一个。
|
||||
|
||||
_pipeUN_ 将会采取另一个预防措施。当父进程结束了等待，父进程将会调用常规的 **exit** 函数去退出。对应的，子进程将会调用 **_exit** 变种来退出，这类变种将快速跟踪终止相关的通知。在效果上，子进程会告诉系统立刻去通知父进程它的这个子进程已经终止了。
|
||||
|
||||
假如两个进程向相同的无名管道中写入内容，字节数据会交错吗？例如，假如进程 P1 向管道写入内容：
|
||||
|
||||
```
|
||||
foo bar
|
||||
```
|
||||
|
||||
同时进程 P2 并发地写入：
|
||||
|
||||
```
|
||||
baz baz
|
||||
```
|
||||
|
||||
到相同的管道，最后的结果似乎是管道中的内容将会是任意错乱的，例如像这样：
|
||||
|
||||
```
|
||||
baz foo baz bar
|
||||
```
|
||||
|
||||
POSIX 标准确保了写不是交错的，使得没有写操作能够超过 **PIPE_BUF** 的范围。在 Linux 系统中， **PIPE_BUF** 的大小是 4096 字节。对于管道我更喜欢只有一个写方和一个读方，从而绕过这个问题。
|
||||
|
||||
## 命名管道
|
||||
|
||||
无名管道没有备份文件：系统将维持一个内存缓存来将字节数据从写方传给读方。一旦写方和读方终止，这个缓存将会被回收，进而无名管道消失。相反的，命名管道有备份文件和一个不同的 API。
|
||||
|
||||
下面让我们通过另一个命令行示例来知晓命名管道的要点。下面是具体的步骤：
|
||||
|
||||
* 开启两个终端。这两个终端的工作目录应该相同。
|
||||
* 在其中一个终端中，键入下面的两个命令（命令行提示符仍然是 **%**，我的注释以 **##** 打头。）：
|
||||
|
||||
```shell
|
||||
% mkfifo tester ## creates a backing file named tester
|
||||
% cat tester ## type the pipe's contents to stdout
|
||||
```
|
||||
|
||||
在最开始，没有任何东西会出现在终端中，因为到现在为止没有命名管道中写入任何东西。
|
||||
* 在第二个终端中输入下面的命令：
|
||||
|
||||
```shell
|
||||
% cat > tester ## redirect keyboard input to the pipe
|
||||
hello, world! ## then hit Return key
|
||||
bye, bye ## ditto
|
||||
<Control-C> ## terminate session with a Control-C
|
||||
```
|
||||
|
||||
无论在这个终端中输入什么，它都会在另一个终端中显示出来。一旦键入 **Ctrl+C**，就会回到正常的命令行提示符，因为管道已经被关闭了。
|
||||
* 通过移除实现命名管道的文件来进行清理：
|
||||
|
||||
```shell
|
||||
% unlink tester
|
||||
```
|
||||
|
||||
正如 _mkfifo_ 程序的名字所暗示的那样，一个命名管道也被叫做一个 FIFO，因为第一个字节先进，然后第一个字节就先出，其他的类似。存在一个名为 **mkfifo** 的库函数，用它可以在程序中创建一个命名管道，它将在下一个示例中被用到，该示例由两个进程组成：一个向命名管道写入，而另一个从该管道读取。
|
||||
|
||||
#### 示例 2. _fifoWriter_ 程序
|
||||
|
||||
```c
|
||||
#include <sys/types.h>
|
||||
#include <sys/stat.h>
|
||||
#include <fcntl.h>
|
||||
#include <unistd.h>
|
||||
#include <time.h>
|
||||
#include <stdlib.h>
|
||||
#include <stdio.h>
|
||||
|
||||
#define MaxLoops 12000 /* outer loop */
|
||||
#define ChunkSize 16 /* how many written at a time */
|
||||
#define IntsPerChunk 4 /* four 4-byte ints per chunk */
|
||||
#define MaxZs 250 /* max microseconds to sleep */
|
||||
|
||||
int main() {
|
||||
const char* pipeName = "./fifoChannel";
|
||||
mkfifo(pipeName, 0666); /* read/write for user/group/others */
|
||||
int fd = open(pipeName, O_CREAT | O_WRONLY); /* open as write-only */
|
||||
if (fd < 0) return -1; /** error **/
|
||||
|
||||
int i;
|
||||
for (i = 0; i < MaxLoops; i++) { /* write MaxWrites times */
|
||||
int j;
|
||||
for (j = 0; j < ChunkSize; j++) { /* each time, write ChunkSize bytes */
|
||||
int k;
|
||||
int chunk[IntsPerChunk];
|
||||
for (k = 0; k < IntsPerChunk; k++)
|
||||
chunk[k] = [rand][9]();
|
||||
write(fd, chunk, sizeof(chunk));
|
||||
}
|
||||
usleep(([rand][9]() % MaxZs) + 1); /* pause a bit for realism */
|
||||
}
|
||||
|
||||
close(fd); /* close pipe: generates an end-of-file */
|
||||
unlink(pipeName); /* unlink from the implementing file */
|
||||
[printf][10]("%i ints sent to the pipe.\n", MaxLoops * ChunkSize * IntsPerChunk);
|
||||
|
||||
return 0;
|
||||
}
|
||||
```
|
||||
|
||||
上面的 _fifoWriter_ 程序可以被总结为如下：
|
||||
|
||||
* 首先程序创建了一个命名管道用来写入数据：
|
||||
|
||||
```c
|
||||
mkfifo(pipeName, 0666); /* read/write perms for user/group/others */
|
||||
int fd = open(pipeName, O_CREAT | O_WRONLY);
|
||||
```
|
||||
|
||||
其中的 **pipeName** 是传递给 **mkfifo** 作为它的第一个参数的备份文件的名字。接着命名管道通过我们熟悉的 **open** 函数调用被打开，而这个函数将会返回一个文件描述符。
|
||||
* 在实现层面上，_fifoWriter_ 不会一次性将所有的数据都写入，而是写入一个块，然后休息随机数目的微秒时间，接着再循环往复。总的来说，有 768000 个 4 比特的整数值被写入到命名管道中。
|
||||
* 在关闭命名管道后，_fifoWriter_ 也将使用 unlink 去掉关联。
|
||||
|
||||
```c
|
||||
close(fd); /* close pipe: generates end-of-stream marker */
|
||||
unlink(pipeName); /* unlink from the implementing file */
|
||||
```
|
||||
|
||||
一旦连接到管道的每个进程都执行了 unlink 操作后，系统将回收这些备份文件。在这个例子中，只有两个这样的进程 _fifoWriter_ 和 _fifoReader_，它们都做了 _unlink_ 操作。
|
||||
|
||||
这个两个程序应该在位于相同工作目录下的不同终端中被执行。但是 _fifoWriter_ 应该在 _fifoReader_ 之前被启动，因为需要 _fifoWriter_ 去创建管道。然后 _fifoReader_ 才能够获取到刚被创建的命名管道。
|
||||
|
||||
#### 示例 3. _fifoReader_ 程序
|
||||
|
||||
```c
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <fcntl.h>
|
||||
#include <unistd.h>
|
||||
|
||||
|
||||
unsigned is_prime(unsigned n) { /* not pretty, but gets the job done efficiently */
|
||||
if (n <= 3) return n > 1;
|
||||
if (0 == (n % 2) || 0 == (n % 3)) return 0;
|
||||
|
||||
unsigned i;
|
||||
for (i = 5; (i * i) <= n; i += 6)
|
||||
if (0 == (n % i) || 0 == (n % (i + 2))) return 0;
|
||||
|
||||
return 1; /* found a prime! */
|
||||
}
|
||||
|
||||
int main() {
|
||||
const char* file = "./fifoChannel";
|
||||
int fd = open(file, O_RDONLY);
|
||||
if (fd < 0) return -1; /* no point in continuing */
|
||||
unsigned count = 0, total = 0, primes_count = 0;
|
||||
|
||||
while (1) {
|
||||
int next;
|
||||
int i;
|
||||
ssize_t count = read(fd, &next, sizeof(int));
|
||||
|
||||
if (0 == count) break; /* end of stream */
|
||||
else if (count == sizeof(int)) { /* read a 4-byte int value */
|
||||
total++;
|
||||
if (is_prime(next)) primes_count++;
|
||||
}
|
||||
}
|
||||
|
||||
close(fd); /* close pipe from read end */
|
||||
unlink(file); /* unlink from the underlying file */
|
||||
[printf][10]("Received ints: %u, primes: %u\n", total, primes_count);
|
||||
|
||||
return 0;
|
||||
}
|
||||
```
|
||||
|
||||
上面的 _fifoReader_ 的内容可以总结为如下：
|
||||
|
||||
* 因为 _fifoWriter_ 已经创建了命名管道，所以 _fifoReader_ 只需要利用标准的 **open** 调用来通过备份文件来获取到管道中的内容：
|
||||
|
||||
```c
|
||||
const char* file = "./fifoChannel";
|
||||
int fd = open(file, O_RDONLY);
|
||||
```
|
||||
|
||||
这个文件的打开是只读的。
|
||||
* 然后这个程序进入一个潜在的无限循环，在每次循环时，尝试读取 4 比特的块。**read** 调用：
|
||||
|
||||
```c
|
||||
ssize_t count = read(fd, &next, sizeof(int));
|
||||
```
|
||||
|
||||
返回 0 来暗示流的结束。在这种情况下，_fifoReader_ 跳出循环，关闭命名管道，并在终止前 unlink 备份文件。
|
||||
* 在读入 4 比特整数后，_fifoReader_ 检查这个数是否为质数。这个操作代表了一个生产级别的读取器可能在接收到的字节数据上执行的逻辑操作。在示例运行中，接收了 768000 个整数中的 37682 个质数。
|
||||
|
||||
在重复的运行示例时， _fifoReader_ 将成功地读取 _fifoWriter_ 写入的所有字节。这不是很让人惊讶的。这两个进程在相同的机器上执行，从而可以不用考虑网络相关的问题。命名管道是一个可信且高效的 IPC 机制，因而被广泛使用。
|
||||
|
||||
下面是这两个程序的输出，在不同的终端中启动，但处于相同的工作目录：
|
||||
|
||||
```shell
|
||||
% ./fifoWriter
|
||||
768000 ints sent to the pipe.
|
||||
###
|
||||
% ./fifoReader
|
||||
Received ints: 768000, primes: 37682
|
||||
```
|
||||
|
||||
### 消息队列
|
||||
|
||||
管道有着严格的先入先出行为：第一个被写入的字节将会第一个被读，第二个写入的字节将第二个被读，以此类推。消息队列可以做出相同的表现，但它又足够灵活，可以使得字节块不以先入先出的次序来接收。
|
||||
|
||||
正如它的名字所建议的那样，消息队列是一系列的消息，每个消息包含两部分：
|
||||
* 荷载，一个字节序列（在 C 中是 **char**）
|
||||
* 一个类型，以一个正整数值的形式给定，类型用来分类消息，为了更灵活的回收
|
||||
|
||||
考虑下面对一个消息队列的描述，每个消息被一个整数类型标记：
|
||||
|
||||
```
|
||||
+-+ +-+ +-+ +-+
|
||||
sender--->|3|--->|2|--->|2|--->|1|--->receiver
|
||||
+-+ +-+ +-+ +-+
|
||||
```
|
||||
|
||||
在上面展示的 4 个消息中，标记为 1 的是开头，即最接近接收端，然后另个标记为 2 的消息，最后接着一个标记为 3 的消息。假如按照严格的 FIFO 行为执行，消息将会以 1-2-2-3 这样的次序被接收。但是消息队列允许其他回收次序。例如，消息可以被接收方以 3-2-1-2 的次序接收。
|
||||
|
||||
_mqueue_ 示例包含两个程序，_sender_ 将向消息队列中写入数据，而 _receiver_ 将从这个队列中读取数据。这两个程序都包含下面展示的头文件 _queue.h_：
|
||||
|
||||
#### 示例 4. 头文件 _queue.h_
|
||||
|
||||
```c
|
||||
#define ProjectId 123
|
||||
#define PathName "queue.h" /* any existing, accessible file would do */
|
||||
#define MsgLen 4
|
||||
#define MsgCount 6
|
||||
|
||||
typedef struct {
|
||||
long type; /* must be of type long */
|
||||
char payload[MsgLen + 1]; /* bytes in the message */
|
||||
} queuedMessage;
|
||||
```
|
||||
|
||||
上面的头文件定义了一个名为 **queuedMessage** 的结构类型，它带有 **payload**（字节数组）和 **type**（整数）这两个域。该文件也定义了一些符号常数（使用 **#define** 语句）。前两个常数被用来生成一个 key，而这个 key 反过来被用来获取一个消息队列的 ID。**ProjectId** 可以是任何正整数值，而 **PathName** 必须是一个存在的，可访问的文件，在这个示例中，指的是文件 _queue.h_。在 _sender_ 和 _receiver_ 中，它们都有的设定语句为：
|
||||
|
||||
```c
|
||||
key_t key = ftok(PathName, ProjectId); /* generate key */
|
||||
int qid = msgget(key, 0666 | IPC_CREAT); /* use key to get queue id */
|
||||
```
|
||||
|
||||
ID **qid** 在效果上是消息队列文件描述符的对应物。
|
||||
|
||||
#### 示例 5. _sender_ 程序
|
||||
|
||||
```c
|
||||
#include <stdio.h>
|
||||
#include <sys/ipc.h>
|
||||
#include <sys/msg.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include "queue.h"
|
||||
|
||||
void report_and_exit(const char* msg) {
|
||||
[perror][6](msg);
|
||||
[exit][7](-1); /* EXIT_FAILURE */
|
||||
}
|
||||
|
||||
int main() {
|
||||
key_t key = ftok(PathName, ProjectId);
|
||||
if (key < 0) report_and_exit("couldn't get key...");
|
||||
|
||||
int qid = msgget(key, 0666 | IPC_CREAT);
|
||||
if (qid < 0) report_and_exit("couldn't get queue id...");
|
||||
|
||||
char* payloads[] = {"msg1", "msg2", "msg3", "msg4", "msg5", "msg6"};
|
||||
int types[] = {1, 1, 2, 2, 3, 3}; /* each must be > 0 */
|
||||
int i;
|
||||
for (i = 0; i < MsgCount; i++) {
|
||||
/* build the message */
|
||||
queuedMessage msg;
|
||||
msg.type = types[i];
|
||||
[strcpy][11](msg.payload, payloads[i]);
|
||||
|
||||
/* send the message */
|
||||
msgsnd(qid, &msg, sizeof(msg), IPC_NOWAIT); /* don't block */
|
||||
[printf][10]("%s sent as type %i\n", msg.payload, (int) msg.type);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
```
|
||||
|
||||
上面的 _sender_ 程序将发送出 6 个消息，每两个为一个类型：前两个是类型 1，接着的连个是类型 2，最后的两个为类型 3。发送的语句：
|
||||
|
||||
```c
|
||||
msgsnd(qid, &msg, sizeof(msg), IPC_NOWAIT);
|
||||
```
|
||||
|
||||
被配置为非阻塞的（**IPC_NOWAIT** 标志），因为这里的消息体量上都很小。唯一的危险在于一个完整的序列将可能导致发送失败，而这个例子不会。下面的 _receiver_ 程序也将使用 **IPC_NOWAIT** 标志来接收消息。
|
||||
|
||||
#### 示例 6. _receiver_ 程序
|
||||
|
||||
```c
|
||||
#include <stdio.h>
|
||||
#include <sys/ipc.h>
|
||||
#include <sys/msg.h>
|
||||
#include <stdlib.h>
|
||||
#include "queue.h"
|
||||
|
||||
void report_and_exit(const char* msg) {
|
||||
[perror][6](msg);
|
||||
[exit][7](-1); /* EXIT_FAILURE */
|
||||
}
|
||||
|
||||
int main() {
|
||||
key_t key= ftok(PathName, ProjectId); /* key to identify the queue */
|
||||
if (key < 0) report_and_exit("key not gotten...");
|
||||
|
||||
int qid = msgget(key, 0666 | IPC_CREAT); /* access if created already */
|
||||
if (qid < 0) report_and_exit("no access to queue...");
|
||||
|
||||
int types[] = {3, 1, 2, 1, 3, 2}; /* different than in sender */
|
||||
int i;
|
||||
for (i = 0; i < MsgCount; i++) {
|
||||
queuedMessage msg; /* defined in queue.h */
|
||||
if (msgrcv(qid, &msg, sizeof(msg), types[i], MSG_NOERROR | IPC_NOWAIT) < 0)
|
||||
[puts][12]("msgrcv trouble...");
|
||||
[printf][10]("%s received as type %i\n", msg.payload, (int) msg.type);
|
||||
}
|
||||
|
||||
/** remove the queue **/
|
||||
if (msgctl(qid, IPC_RMID, NULL) < 0) /* NULL = 'no flags' */
|
||||
report_and_exit("trouble removing queue...");
|
||||
|
||||
return 0;
|
||||
}
|
||||
```
|
||||
|
||||
这个 _receiver_ 程序不会创建消息队列，尽管 API 看起来像是那样。在 _receiver_ 中，对
|
||||
|
||||
```c
|
||||
int qid = msgget(key, 0666 | IPC_CREAT);
|
||||
```
|
||||
|
||||
的调用可能因为带有 **IPC_CREAT** 标志而具有误导性，但是这个标志的真实意义是 _如果需要就创建，否则直接获取_。_sender_ 程序调用 **msgsnd** 来发送消息，而 _receiver_ 调用 **msgrcv** 来接收它们。在这个例子中，_sender_ 以 1-1-2-2-3-3 的次序发送消息，但 _receiver_ 接收它们的次序为 3-1-2-1-3-2，这显示消息队列没有被严格的 FIFO 行为所拘泥：
|
||||
|
||||
```shell
|
||||
% ./sender
|
||||
msg1 sent as type 1
|
||||
msg2 sent as type 1
|
||||
msg3 sent as type 2
|
||||
msg4 sent as type 2
|
||||
msg5 sent as type 3
|
||||
msg6 sent as type 3
|
||||
|
||||
% ./receiver
|
||||
msg5 received as type 3
|
||||
msg1 received as type 1
|
||||
msg3 received as type 2
|
||||
msg2 received as type 1
|
||||
msg6 received as type 3
|
||||
msg4 received as type 2
|
||||
```
|
||||
|
||||
上面的输出显示 _sender_ 和 _receiver_ 可以在同一个终端中启动。输出也显示消息队列是持久的，即便在 _sender_ 进程在完成创建队列，向队列写数据，然后离开的整个过程后，队列仍然存在。只有在 _receiver_ 进程显式地调用 **msgctl** 来移除该队列，这个队列才会消失：
|
||||
|
||||
```c
|
||||
if (msgctl(qid, IPC_RMID, NULL) < 0) /* remove queue */
|
||||
```
|
||||
|
||||
### 总结
|
||||
|
||||
管道和消息队列的 API 在根本上来说都是单向的：一个进程写，然后另一个进程读。当然还存在双向命名管道的实现，但我认为这个 IPC 机制在它最为简单的时候反而是最佳的。正如前面提到的那样，消息队列已经不大受欢迎了，尽管没有找到什么特别好的原因来解释这个现象。而队列仍然是 IPC 工具箱中的另一个工具。这个快速的 IPC 工具箱之旅将以第 3 部分-通过套接字和信号来示例 IPC -来终结。
|
||||
|
||||
--------------------------------------------------------------------------------
|
||||
|
||||
via: https://opensource.com/article/19/4/interprocess-communication-linux-channels
|
||||
|
||||
作者：[Marty Kalin][a]
|
||||
选题：[lujun9972][b]
|
||||
译者：[FSSlc](https://github.com/FSSlc)
|
||||
校对：[校对者ID](https://github.com/校对者ID)
|
||||
|
||||
本文由 [LCTT](https://github.com/LCTT/TranslateProject) 原创编译，[Linux中国](https://linux.cn/) 荣誉推出
|
||||
|
||||
[a]: https://opensource.com/users/mkalindepauledu
|
||||
[b]: https://github.com/lujun9972
|
||||
[1]: https://opensource.com/sites/default/files/styles/image-full-size/public/lead-images/talk_chat_communication_team.png?itok=CYfZ_gE7 (Chat bubbles)
|
||||
[2]: https://en.wikipedia.org/wiki/Inter-process_communication
|
||||
[3]: https://opensource.com/article/19/4/interprocess-communication-ipc-linux-part-1
|
||||
[4]: http://man7.org/linux/man-pages/man2/mq_open.2.html
|
||||
[5]: http://man7.org/linux/man-pages/man2/mq_open.2.html#ATTRIBUTES
|
||||
[6]: http://www.opengroup.org/onlinepubs/009695399/functions/perror.html
|
||||
[7]: http://www.opengroup.org/onlinepubs/009695399/functions/exit.html
|
||||
[8]: http://www.opengroup.org/onlinepubs/009695399/functions/strlen.html
|
||||
[9]: http://www.opengroup.org/onlinepubs/009695399/functions/rand.html
|
||||
[10]: http://www.opengroup.org/onlinepubs/009695399/functions/printf.html
|
||||
[11]: http://www.opengroup.org/onlinepubs/009695399/functions/strcpy.html
|
||||
[12]: http://www.opengroup.org/onlinepubs/009695399/functions/puts.html
|
||||
Loading…
Reference in New Issue
Block a user