Linux進程間通信—

1.信號的介紹

????????信號( Signal )是 Unix, 類Unix以及其他POSIX兼容的操作系統中進程間通信的一種有限制的手段。

1）信號是在軟件層面上對中斷機制的一種模擬，是一種異步通信方式。
2）信號可以直接進行用戶空間進程和內核進程之間的交互，內核進程也可以利用呀來通知用戶空間進程發生了那些系統事件——用來提醒進程一個事件已經發生，當一個信號發送給一個進程，操作系統中斷了進程正常的控制流程），此時，任何非原子操作都將被中斷。
3）如果該進程當前未處于執行態，則該信號就由內核保存起來，直到該進程恢復執行再傳遞給它；如果一個信號被進程設置為阻塞，則該信號的傳遞被延遲，直到其阻塞被取消時才傳遞給進程

? ? ? ? 在linux系統中，提供了幾十種信號，分別代表不同的意義。通過下面的bash命令，可以得到Linux中各種信號，供64種：

~$ kill -l

? ? ? ? 信號可以在任何時候發送給某一個進程，進程需要為這個信號配置信號處理函數嗎。當某個信號發生時，就默認執行這個函數就可以。

????????通過以下指令可以查看各個信號的具體含義和對應的處理方法

man 7 signal

? ? ??

? 由上表可見，信號的處理大致分為三種：

執行默認操作(default action)。Linux對每種信號都規定了默認操作，例如上面列表中的Action 列的Term, Core等：
- Term : “ Default action is to terminate the process ”就是終止進程的默認操作
- Ign：“ Default action is to ignore the signal "就是忽略進程的默認操作
- Core：” Default action is to terminate the process and dump core (see core(5)) “就是終止進程并生成 core dump 文件（通常用于調試）的默認操作
- Stop : " Default action is to stop the process "就是停止進程的默認操作
- Cont : " Default action is to continue the process if it is currently stopped "就是恢復執行已被停止的進程的默認操作
捕捉信號。我們可以為信號定義一個信號處理函數，當信號發生時，我們就執行相應的信號處理函數。
忽略信號。當我們不希望處理某些信號的時候，就可以忽略該信號，不做任何處理。有兩個信號是應用進程無法捕捉和忽略的，即SIGKILL 和 SEGSTOP，它們用于在任何時候中斷或結束某一進程。

??????????????????????????????????????

2.信號的來源

信號事件的來源有兩種：

硬件來源（本質是對硬件中斷的一種模擬）
軟件來源，最常用發送信號的系統函數是 kill, alarm和settimer以及sigqueue函數，軟件來源還包括一些非法運算等操作

????????信號可以在直接與用戶空間進程和內核進程之間的交互，內核進程也可以利用它來通知用戶空間進程發生了哪些系統事件。
????????如果該進程當前未處于執行態，則該信號就由內核保存起來，直到該進程恢復執行再傳遞給它；如果一個信號被進程設置為阻塞，則該進程的傳遞就被延遲，直到其阻塞取消時才被傳遞給進程。

3.信號與中斷的異同

相似之處：

均是異步通信方式
均會注冊處理函數，都是用于對當前的任務進行一些處理，如調度，停止等。當檢測出有信號或中斷請求時，都會暫停當前任務而轉去執行相應的處理程序。
在處理程序結束后，都會返回斷點
信號和中斷都可進行屏蔽

但是二者實際上有很大不同，其主要區別有：

中斷和信號雖然都可能源于硬件和軟件，但是中斷處理函數注冊于內核之中，由內核中運行，而信號的處理函數注冊于用戶態，內核收到信號后根據當前任務task_struct結構體中的信號相關的數據結構找尋對應的處理函數并最終在用戶態處理
中斷作用于內核全局，而信號作用于當前任務（進程）。即信號影響的往往是一個進程，而中斷處理如果出現問題則會導致整個Linux內核的崩潰。
中斷是有優先級的，而信號不存在，信號之間都是“平等”的；
中斷是及時響的，而信號具有很大的時延

????????既然信號是中斷的一種模擬，那為什么不直接采用中斷而是設計了信號呢？

? ? ? ? （1）安全性：由于中斷的異常處理可能會導致內核崩潰，所以直接操作會維系餓系統穩定和安全

? ? ? ? （2）簡化：信號將復雜的硬件中斷機制抽象成簡單事件通知，用戶進程只需關心”發生了什么“，不關心”中斷源是如何處理的“

? ? ? ? （3）靈活性：信號可進程捕獲，忽略或自定義處理，而中斷相應必須立即，原子地完成，不允許復雜的用戶處理邏輯

4.注冊信號處理函數

????????有些時候我們希望能夠讓信號運行一些特殊處理功能，所以有了自定義的信號處理函數。注冊API主要有signal()和sigaction() （比較推薦）。

/* Set the handler for the signal SIG to HANDLER,returning the old handler, or SIG_ERR on error.  */
__sighandler_t
signal (int sig, __sighandler_t handler)
{__set_errno (ENOSYS);return SIG_ERR;
}

? ? ? ? 其主要區別在于sigaction()對于信號sig會綁定對應的結構體sigaction而不僅僅是一個處理函數sighandler_t。這樣做的好處是可以更精細地控制信號處理，通過不同參數實現不同的效果。例如sa_flags可以設置如：

SA_ONESHOT：信號處理函數僅作用一次，之后啟用默認行為
SA_NOMASK：該信號處理函數執行過程中允許被其他信號或者相同信號中斷，即不屏蔽
SA_INTERRUPT：該信號處理函數若執行過程中被中斷，則不會再調度回該函數繼續執行，而是直接返回-EINTR，將執行邏輯交還給調用方
SA_RESTART：與SA_INTERRUPT相反，會自動重啟該函數

/* Structure describing the action to be taken when a signal arrives.  */
struct sigaction{/* Signal handler.  */
#if defined __USE_POSIX199309 || defined __USE_XOPEN_EXTENDEDunion{/* Used if SA_SIGINFO is not set.  */__sighandler_t sa_handler;/* Used if SA_SIGINFO is set.  */void (*sa_sigaction) (int, siginfo_t *, void *);}__sigaction_handler;
# define sa_handler	__sigaction_handler.sa_handler
# define sa_sigaction	__sigaction_handler.sa_sigaction
#else__sighandler_t sa_handler;
#endif/* Additional set of signals to be blocked.  */__sigset_t sa_mask;/* Special flags.  */int sa_flags;};

4.1 用戶調用

????????通常用戶會在代碼中這樣編寫

struct sigaction act;
act.sa_handler = my_handler; // 或 sa_sigaction = my_sigaction
sigemptyset(&act.sa_mask);
act.sa_flags = 0;
sigaction(SIGINT, &act, NULL);

4.2 進入glibc層

????????用戶在代碼中調用的 sigaction本質上就是 __sigaction()

/* If ACT is not NULL, change the action for SIG to *ACT.If OACT is not NULL, put the old action for SIG in *OACT.  */
int
__sigaction (int sig, const struct sigaction *act, struct sigaction *oact)
{if (sig <= 0 || sig >= NSIG || is_internal_signal (sig)){__set_errno (EINVAL);return -1;}internal_sigset_t set;if (sig == SIGABRT)__abort_lock_wrlock (&set);int r = __libc_sigaction (sig, act, oact);if (sig == SIGABRT)__abort_lock_unlock (&set);return r;
}

? ? ? ? 首先進行合法性檢查，再對sig == SIGABRT的特殊信號進行加鎖處理，然后調用__libc_sigaction(sig, act, oact)，通常最終系統調用為rt_sigacton().

? ? ? ? rt_sigaction()是Linux內核對信號注冊的實現，內核會維護每個進程的信號處理表

如果act非空，內核會將信號編號對應的handler設置為你傳入的處理函數，同時記錄掩碼和flags等
如果oact非空，內核會將原先的處理方式返回給用戶

SYSCALL_DEFINE4(rt_sigaction, int, sig,const struct sigaction __user *, act,struct sigaction __user *, oact,size_t, sigsetsize)
{struct k_sigaction new_sa, old_sa;int ret = -EINVAL;
......if (act) {if (copy_from_user(&new_sa.sa, act, sizeof(new_sa.sa)))return -EFAULT;}ret = do_sigaction(sig, act ? &new_sa : NULL, oact ? &old_sa : NULL);if (!ret && oact) {if (copy_to_user(oact, &old_sa.sa, sizeof(old_sa.sa)))return -EFAULT;}
out:return ret;
}

? ? ? ? do_sigaction()會將用戶層傳來的信號處理函數賦值給當前任務task_struct_current對應的sighand->action[]數組中sig信號對應的位置，以用于之后調用。

int do_sigaction(int sig, struct k_sigaction *act, struct k_sigaction *oact)
{struct task_struct *p = current, *t;struct k_sigaction *k;sigset_t mask;
......k = &p->sighand->action[sig-1];spin_lock_irq(&p->sighand->siglock);if (oact)*oact = *k;if (act) {sigdelsetmask(&act->sa.sa_mask, sigmask(SIGKILL) | sigmask(SIGSTOP));*k = *act;
......}spin_unlock_irq(&p->sighand->siglock);return 0;
}

5.發送信號

????????不論通過kiil還是sigqueue系統調用還是通過tkill或者tgkill發送指定線程的信號，其最終調用的均是do_send_sig_info()函數，其調用鏈如下所示：

kill()->kill_something_info()->kill_pid_info()->group_send_sig_info()->do_send_sig_info()tkill()->do_tkill()->do_send_specific()->do_send_sig_info()tgkill()->do_tkill()->do_send_specific()->do_send_sig_info()rt_sigqueueinfo()->do_rt_sigqueueinfo()->kill_proc_info()->kill_pid_info()->group_send_sig_info()->do_send_sig_info()

do_send_sig_info() 會調用 send_signal()，進而調用 __send_signal()。這里代碼比較復雜，主要邏輯如下

根據發送信號的類型判斷是共享信號還是線程獨享信號，由此賦值pending。如果是 kill 發送的，也就是發送給整個進程的，就應該發送給 t->signal->shared_pending，這里面是整個進程所有線程共享的信號；如果是 tkill 發送的，也就是發給某個線程的，就應該發給 t->pending，這里面是這個線程的 task_struct 獨享的。
調用 legacy_queue()判斷是否為可靠信號，不可靠則直接退出
調用__sigqueue_alloc() 分配一個 struct sigqueue 對象，然后通過 list_add_tail 掛在 struct sigpending 里面的鏈表上。
調用 complete_signal()分配線程處理該信號

5.1信號的排隊和丟失

不可靠信號（傳統信號，編號 < SIGRTMIN，通常為1~31）

通過legacy_queue()判斷：如果該信號已存在于目標的信號集合（sigpending）中，則直接丟棄新來的同類型信號（不重復排隊）。
這樣做的后果是：同一個不可靠信號只會被投遞一次，多次發送會丟失。
這也是“不可靠信號”的核心定義（不是說系統不可靠，而是信號可能丟失）。

可靠信號（實時信號，編號 ≥ SIGRTMIN，通常為32~64）

每一個實時信號都可以排隊，即多次發送會在隊列中保存多份，不會丟失。
通過__sigqueue_alloc()分配sigqueue對象，鏈入待處理隊列。

static inline int legacy_queue(struct sigpending *signals, int sig)
{return (sig < SIGRTMIN) && sigismember(&signals->signal, sig);
}#define SIGRTMIN  32
#define SIGRTMAX  _NSIG
#define _NSIG    64

5.2信號投遞與喚醒過程

信號（無論可靠、不可靠）加入進程/線程的sigpending或shared_pending集合/隊列。

調用complete_signal()尋找適合的線程處理該信號：

優先找主線程。
多線程時，遍歷線程組，找到合適的線程喚醒。
如果是致命信號（如SIGKILL），則全組強制喚醒并終止。

static void complete_signal(int sig, struct task_struct *p, enum pid_type type)
{struct signal_struct *signal = p->signal;struct task_struct *t;/** Now find a thread we can wake up to take the signal off the queue.** If the main thread wants the signal, it gets first crack.* Probably the least surprising to the average bear.*/if (wants_signal(sig, p))t = p;else if ((type == PIDTYPE_PID) || thread_group_empty(p))/** There is just one thread and it does not need to be woken.* It will dequeue unblocked signals before it runs again.*/return;else {/** Otherwise try to find a suitable thread.*/t = signal->curr_target;while (!wants_signal(sig, t)) {t = next_thread(t);if (t == signal->curr_target)/** No thread needs to be woken.* Any eligible threads will see* the signal in the queue soon.*/return;}signal->curr_target = t;}/** Found a killable thread.  If the signal will be fatal,* then start taking the whole group down immediately.*/if (sig_fatal(p, sig) &&!(signal->flags & SIGNAL_GROUP_EXIT) &&!sigismember(&t->real_blocked, sig) &&(sig == SIGKILL || !p->ptrace)) {/** This signal will be fatal to the whole group.*/if (!sig_kernel_coredump(sig)) {/** Start a group exit and wake everybody up.* This way we don't have other threads* running and doing things after a slower* thread has the fatal signal pending.*/signal->flags = SIGNAL_GROUP_EXIT;signal->group_exit_code = sig;signal->group_stop_count = 0;t = p;do {task_clear_jobctl_pending(t, JOBCTL_PENDING_MASK);sigaddset(&t->pending.signal, SIGKILL);signal_wake_up(t, 1);} while_each_thread(p, t);return;}}/** The signal is already in the shared-pending queue.* Tell the chosen thread to wake up and dequeue it.*/signal_wake_up(t, sig == SIGKILL);return;
}

喚醒的關鍵是調用signal_wake_up()：

設置線程的TIF_SIGPENDING標志，表示“有信號待處理”。
嘗試喚醒目標線程（如果已在TASK_RUNNING狀態，則通過IPI強制調度）。

static inline void signal_wake_up(struct task_struct *t, bool resume)
{signal_wake_up_state(t, resume ? TASK_WAKEKILL : 0);
}void signal_wake_up_state(struct task_struct *t, unsigned int state)
{set_tsk_thread_flag(t, TIF_SIGPENDING);/** TASK_WAKEKILL also means wake it up in the stopped/traced/killable* case. We don't check t->state here because there is a race with it* executing another processor and just now entering stopped state.* By using wake_up_state, we ensure the process will wake up and* handle its death signal.*/if (!wake_up_state(t, state | TASK_INTERRUPTIBLE))kick_process(t);
}

6. 信號的處理

????????這里我們以一個從tap 網卡中讀取數據的例子來分析信號的處理邏輯。從網卡讀取數據會通過系統調用進入內核，之后通過函數調用表找到對應的函數執行。在讀的過程中，如果沒有數據處理則會調用schedule()函數主動讓出CPU進入休眠狀態并等待再次喚醒。

???????

tap_do_read()主要邏輯為：

把當前進程或者線程的狀態設置為 TASK_INTERRUPTIBLE，這樣才能使這個系統調用可以被中斷。
可以被中斷的系統調用往往是比較慢的調用，并且會因為數據不就緒而通過 schedule() 讓出 CPU 進入等待狀態。在發送信號的時候，我們除了設置這個進程和線程的 _TIF_SIGPENDING 標識位之外，還試圖喚醒這個進程或者線程，也就是將它從等待狀態中設置為 TASK_RUNNING。當這個進程或者線程再次運行的時候，會從 schedule() 函數中返回，然后再次進入 while 循環。由于這個進程或者線程是由信號喚醒的而不是因為數據來了而喚醒的，因而是讀不到數據的，但是在 signal_pending() 函數中，我們檢測到了 _TIF_SIGPENDING 標識位，這說明系統調用沒有真的做完，于是返回一個錯誤 ERESTARTSYS，然后帶著這個錯誤從系統調用返回。
如果沒有信號，則繼續調用schedule()讓出CPU

static ssize_t tap_do_read(struct tap_queue *q,struct iov_iter *to,int noblock, struct sk_buff *skb)
{DEFINE_WAIT(wait);ssize_t ret = 0;if (!iov_iter_count(to)) {kfree_skb(skb);return 0;}if (skb)goto put;while (1) {if (!noblock)prepare_to_wait(sk_sleep(&q->sk), &wait,TASK_INTERRUPTIBLE);/* Read frames from the queue */skb = ptr_ring_consume(&q->ring);if (skb)break;if (noblock) {ret = -EAGAIN;break;}if (signal_pending(current)) {ret = -ERESTARTSYS;break;}/* Nothing to read, let's sleep */schedule();}if (!noblock)finish_wait(sk_sleep(&q->sk), &wait);put:if (skb) {ret = tap_put_user(q, skb, to);if (unlikely(ret < 0))kfree_skb(skb);elseconsume_skb(skb);}return ret;
}

????????schedule()會在系統調用負擔會或者中斷返回的時刻調用exit_to_usermode_loop()，在任務調度中的標記位為_TIF_NEED_RESCHED，而對于信號來說就是_TIF_SIGPENDING。

static void exit_to_usermode_loop(struct pt_regs *regs, u32 cached_flags)
{while (true) {
......if (cached_flags & _TIF_NEED_RESCHED)schedule();
....../* deal with pending signal delivery */if (cached_flags & _TIF_SIGPENDING)do_signal(regs);
......if (!(cached_flags & EXIT_TO_USERMODE_LOOP_FLAGS))break;}
}

? ? ? ? do_signal()函數會調用handle_signal()函數，這里主要存在一個問題使得邏輯變得較為復雜：信號處理函數定義于用戶態，而調度過程位于內核態。

/** Note that 'init' is a special process: it doesn't get signals it doesn't* want to handle. Thus you cannot kill init even with a SIGKILL even by* mistake.*/
void do_signal(struct pt_regs *regs)
{struct ksignal ksig;if (get_signal(&ksig)) {/* Whee! Actually deliver the signal.  */handle_signal(&ksig, regs);return;}/* Did we come from a system call? */if (syscall_get_nr(current, regs) >= 0) {/* Restart the system call - no handlers present */switch (syscall_get_error(current, regs)) {case -ERESTARTNOHAND:case -ERESTARTSYS:case -ERESTARTNOINTR:regs->ax = regs->orig_ax;regs->ip -= 2;break;case -ERESTART_RESTARTBLOCK:regs->ax = get_nr_restart_syscall(regs);regs->ip -= 2;break;}}/** If there's no signal to deliver, we just put the saved sigmask* back.*/restore_saved_sigmask();
}

????????handle_signal()會判斷當前是否從系統調用調度而來，當發現錯誤碼為ERESTARTSYS的時候就知道這是從一個沒有調用完的系統調用返回的，設置系統錯誤碼為EINTR。由于此處不會直接返回任務調度前記錄的用戶態狀態，而是進入注冊好的信號處理函數，因此需要調用setup_rt_frame()構建新的寄存器結構體pt_regs。

static void
handle_signal(struct ksignal *ksig, struct pt_regs *regs)
{bool stepping, failed;
....../* Are we from a system call? */if (syscall_get_nr(current, regs) >= 0) {/* If so, check system call restarting.. */switch (syscall_get_error(current, regs)) {case -ERESTART_RESTARTBLOCK:case -ERESTARTNOHAND:regs->ax = -EINTR;break;case -ERESTARTSYS:if (!(ksig->ka.sa.sa_flags & SA_RESTART)) {regs->ax = -EINTR;break;}/* fallthrough */case -ERESTARTNOINTR:regs->ax = regs->orig_ax;regs->ip -= 2;break;}}
......failed = (setup_rt_frame(ksig, regs) < 0);
......signal

setup_rt_frame()主要調用__setup_rt_frame()，主要邏輯為：

調用get_sigframe()得到regs中的sp寄存器值，即原進程用戶態的棧頂指針，將sp減去sizeof(struct rt_sigframe)從而把該新建棧幀壓入棧
調用put_user_ex()，將 sa_restorer 按照函數棧的規則放到了 frame->pretcode 里面。函數棧里面包含了函數執行完跳回去的地址,當 sa_handler 執行完之后，彈出的函數棧是 frame，也就應該跳到 sa_restorer 的地址
調用setup_sigcontext() 里面，將原來的 pt_regs 保存在了 frame 中的 uc_mcontext 里
填充regs，將regs->ip設置為自定義的信號處理函數sa_handler，將棧頂regs->sp設置為新棧幀frame地址

static int
setup_rt_frame(struct ksignal *ksig, struct pt_regs *regs)
{
......return __setup_rt_frame(ksig->sig, ksig, set, regs);
......
}static int __setup_rt_frame(int sig, struct ksignal *ksig,sigset_t *set, struct pt_regs *regs)
{struct rt_sigframe __user *frame;void __user *fp = NULL;int err = 0;frame = get_sigframe(&ksig->ka, regs, sizeof(struct rt_sigframe), &fp);
......put_user_try {
....../* Set up to return from userspace.  If provided, use a stubalready in userspace.  *//* x86-64 should always use SA_RESTORER. */if (ksig->ka.sa.sa_flags & SA_RESTORER) {put_user_ex(ksig->ka.sa.sa_restorer, &frame->pretcode);} else {/* could use a vstub here */err |= -EFAULT;}} put_user_catch(err);err |= setup_sigcontext(&frame->uc.uc_mcontext, fp, regs, set->sig[0]);err |= __copy_to_user(&frame->uc.uc_sigmask, set, sizeof(*set));if (err)return -EFAULT;/* Set up registers for signal handler */regs->di = sig;/* In case the signal handler was declared without prototypes */regs->ax = 0;/* This also works for non SA_SIGINFO handlers because they expect thenext argument after the signal number on the stack. */regs->si = (unsigned long)&frame->info;regs->dx = (unsigned long)&frame->uc;regs->ip = (unsigned long) ksig->ka.sa.sa_handler;regs->sp = (unsigned long)frame;/** Set up the CS and SS registers to run signal handlers in* 64-bit mode, even if the handler happens to be interrupting* 32-bit or 16-bit code.** SS is subtle.  In 64-bit mode, we don't need any particular* SS descriptor, but we do need SS to be valid.  It's possible* that the old SS is entirely bogus -- this can happen if the* signal we're trying to deliver is #GP or #SS caused by a bad* SS value.  We also have a compatbility issue here: DOSEMU* relies on the contents of the SS register indicating the* SS value at the time of the signal, even though that code in* DOSEMU predates sigreturn's ability to restore SS.  (DOSEMU* avoids relying on sigreturn to restore SS; instead it uses* a trampoline.)  So we do our best: if the old SS was valid,* we keep it.  Otherwise we replace it.*/regs->cs = __USER_CS;if (unlikely(regs->ss != __USER_DS))force_valid_ss(regs);return 0;
}

7.信號的發送、處理總結

假設我們有一個進程A會從tap網卡中讀取數據，main函數里面調用系統調用通過中斷陷入內核
按照系統調用原理，將用戶態棧的信息保存在 pt_regs 里面，也即記住原來用戶態是運行到了 line A 的位置
在內核中執行系統調用讀取數據
當發現沒有什么數據可讀取的時候進入睡眠狀態，并且調用schedule()讓出CPU
將進程狀態設置為可中斷的睡眠狀態TASK_INTERRUPTIBLE，也即如果有信號來的話是可以喚醒它
其他的進程或者shell通過調用kill(), tkill(), tgkill(),rt_sigqueueinfo()發送信號。四個發送信號的函數，在內核中最終調用都是do_send_sig_info()
do_send_sig_info()調用send_signal()給進程A發送一個信號，其實就是找到進程A的task_struct，不可靠信號加入信號集合，可靠信號加入信號鏈表
do_send_ sig_info()調用signal_wake_up()喚醒進程A
進程A重新進入運行狀態TASK_RUNNING，接著schedule()運行
進程A被喚醒后檢查是否有信號到達，如果沒有，重新循環到一開始，嘗試再次讀取數據，如果還是沒有數據，再次進入TASK_INTERRUPTIBLE，即可中斷的睡眠狀態
當發現有信號到來時，就返回當前正在執行的系統調用，并返回一個錯誤表示系統調用被中斷了
系統調用返回時，會調用exit_to_usermode_loop()，這是一個信號處理的時機
調用do_signal()開始處理信號
根據信號得到信號處理函數Sa_handler，然后修改pt_regs中的用戶態棧的信息讓 pt_regs指向 sa_handler，同時修改用戶態的棧，插入一個棧幀sa_restorer,里面保存了原來的指向line A的pt_regs，并且設置讓sa_handler運行完畢時跳到sa_restorer運行
返回用戶態，由于pt_regs已經設置為sa_handler,則返回用戶態執行sa_handler
sa_restorer會調用系統調用rt_sigreturn再次進入內核
在內核中，rt_sigreturn恢復到原來的pt_regs，重新指向line A
從rt_sigreturn返回用戶態，還是調用exit_to_usermode_loop()
這次因為pt_regs已經指向line A了，于是就到了進程A中接著系統調用之后運行，當然這個系統調用返回的是它被中斷了沒有執行完的錯誤