分类 C 下的文章

Context

作者: nbboy
时间: 2021-05-10
分类: 软件架构,设计模式,C,Golang
评论

# Context
> 使用上下文的一个很好的心理模型是它应该在您的程序中流通，想象一条河或流水。

**分析版本:1.15.2**

### Context

context接口只有4个方法,先看下接口定义

```go
type Context interface {
   //超时时间
   Deadline() (deadline time.Time, ok bool)
  
   //需要监听的通道
   Done() <-chan struct{}
  
   //如果没有关闭，则返回nil,否则返回一个错误值
   Err() error
  
   //指定key的value
   Value(key interface{}) interface{}
}
```

目前版本中，实现的struct为emptyCtx,cancelCtx,timerCtx,valueCtx,每个ctx对应的应用场景都不一样,先看下最简单的emptyCtx

### emptyCtx

```go
var (
   background = new(emptyCtx)
   todo       = new(emptyCtx)
)

func Background() Context {
   return background
}

func TODO() Context {
   return todo
}
```

TODO和Background用的都是emptyCtx,Background主要被用来作为其他Ctx的根,而TODO主要可以视为一种nil的Ctx去用，因为在官方的设计中，不允许使用nil作为Ctx的值。emptyCtx的实现非常简单，不做具体介绍，都是空的方法体。

先看下比较常用的cancelCtx的使用方法

### cancelCtx

```go
gen := func(ctx context.Context) <-chan int {
   dst := make(chan int)
   n := 1
   go func() {
      for {
         select {
         case <-ctx.Done()://取消后从这里返回
            fmt.Println("ctx.done")
            return
         case dst <- n:
            n++
         }
      }
   }()
   return dst
}

ctx, cancel := context.WithCancel(context.Background())

for n := range gen(ctx) {
   fmt.Println(n)
   if n == 5 {
      //达到目标，取消ctx
      cancel()
      break
   }
}

time.Sleep(3 * time.Second)
```

cancelCtx主要用来控制goroutine的生命周期,即什么时候结束生命周期,当然这个需要goroutine本身去配合,select Done返回的通道。再看下，cancelCtx的内部结构

```go
type cancelCtx struct {
   Context

mu       sync.Mutex            // protects following fields 用来保护成员
   done     chan struct{}         // created lazily, closed by first cancel call 就是Done()返回的chan，调用cancel()后就被关闭
   children map[canceler]struct{} // set to nil by the first cancel call 子ctx,所有ctx会组成一颗树形结构,而此处指向其孩子节点
   err      error                 // set to non-nil by the first cancel call 调用cancel()后，被设置成取消原因
}
```

这里看下cancelCtx的构建函数

```go
func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
   if parent == nil {
      panic("cannot create context from nil parent")
   }
  
   //创建cancelCtx,关联传递进来的ctx
   c := newCancelCtx(parent)

propagateCancel(parent, &c)
   return &c, func() { c.cancel(true, Canceled) }
}

func propagateCancel(parent Context, child canceler) {
	//首先检查父ctx是否关闭
	done := parent.Done()
	if done == nil {
		return // parent is never canceled
	}

select {
	case <-done:
		// parent is already canceled
		// 如果父ctx被取消，则也同步取消子ctx
		child.cancel(false, parent.Err())
		return
	default:
	}

//如果找到了cancelCtx
	if p, ok := parentCancelCtx(parent); ok {
		p.mu.Lock()
		if p.err != nil {//父节点已经被取消
			// parent has already been canceled
			child.cancel(false, p.err)
		} else {//如果没被取消，则把子节点挂到父节点
			if p.children == nil {
				p.children = make(map[canceler]struct{})
			}
			p.children[child] = struct{}{}
		}
		p.mu.Unlock()
	} else {
		//如果是自定义的ctx,就会开启一个goroutine去监听父的取消事件，并且取消子ctx
		atomic.AddInt32(&goroutines, +1)
		go func() {
			select {
			case <-parent.Done():
				child.cancel(false, parent.Err())
			case <-child.Done():
			}
		}()
	}
}

func parentCancelCtx(parent Context) (*cancelCtx, bool) {
	done := parent.Done()

//如果关闭，则返回false
	if done == closedchan || done == nil {
		return nil, false
	}

//看下是否cancelCtx
	p, ok := parent.Value(&cancelCtxKey).(*cancelCtx)

//没找到则返回false
	if !ok {
		return nil, false
	}
	p.mu.Lock()
	ok = p.done == done
	p.mu.Unlock()

//如果不一样也返回flse
	if !ok {
		return nil, false
	}

//通过深层查找，找到了cancelCxt,则才返回true
	return p, true
}
```

- 父Ctx如果不需要被取消，则直接返回，Background,TODO就是不需要被取消的类型

- 如果父ctx被取消，则也同步取消子ctx

- parentCancelCtx会深层次得去找父cancelCtx,这里分两种情况

1）如果是标准(cancelCtx,timerCtx)则会同步父子Ctx的状态（要么都同步取消，要么建立关系）

2）如果是自定义Ctx,就会开启一个goroutine去监听父的取消事件，并且取消子ctx

这里这么做的原因，就是需要把子节点的状态和父节点要同步,调用withCancel()返回的cancel函数其实是调用cancelCtx.cancel()函数

```go
// cancel closes c.done, cancels each of c's children, and, if
// removeFromParent is true, removes c from its parent's children.
func (c *cancelCtx) cancel(removeFromParent bool, err error) {
   if err == nil {
      panic("context: internal error: missing cancel error")
   }
   c.mu.Lock()
   if c.err != nil {
      c.mu.Unlock()
      return // already canceled
   }
   c.err = err
   if c.done == nil {
      c.done = closedchan
   } else {
      //关闭chan
      close(c.done)
   }

//子ctx依次进行cancel
   for child := range c.children {
      // NOTE: acquiring the child's lock while holding parent's lock.
      child.cancel(false, err)
   }
   c.children = nil
   c.mu.Unlock()

//从根节点里移除c
   if removeFromParent {
      removeChild(c.Context, c)
   }
}
```

注释中说的很明白，会把打开的chan关闭，然后依次调用子ctx的cancel,所以如果我们忘记调用cancel,其实会有大量的chan没被close掉,然后造成资源的浪费！

此处试下级联取消,代码如下

```go
func main() {
   ctx1, cancel1 := context.WithCancel(context.Background())
   defer cancel1()

ctx2, cancel2 := context.WithCancel(ctx1)
   defer cancel2()

ctx3, cancel3 := context.WithCancel(ctx2)
   defer cancel3()

go func() {
      select {
      case <-time.After(3 * time.Second):
         cancel1()
      }
   }()

<-ctx3.Done()
}
```

创建了三个ctx,然后第一个ctx取消后,其下的所有ctx都会取消。需要注意，代码中其实ctx1被cancel了两次，通过了解实现的代码，知道这么写其实并没有什么问题。画一个图，直观了解下3个ctx组成的结构。取消是沿着继承链，从除了根部外（Background不能被取消）一直到所有节点执行取消操作!

┌───────────────────┐
           │ ┌───────────────┐ │
           │ │  Background   │ │
           │ └───────────────┘ │
           └─────────┬─────────┘
                     │
                     │
                     │
                     │
       │   ┌─────────▼─────────┐
       │   │ ┌───────────────┐ │
       │   │ │     Ctx1      │ │
       │   │ └───────────────┘ │
       │   └─────────┬─────────┘
       │             │
       │             │
       │             │
       │             │
       │   ┌─────────▼─────────┐
       │   │ ┌───────────────┐ │
    Cancel │ │     Ctx2      │ │
       │   │ └───────────────┘ │
       │   └─────────┬─────────┘
       │             │
       │             │
       │             │
       │             │
       │   ┌─────────▼─────────┐
       │   │ ┌───────────────┐ │
       │   │ │     Ctx3      │ │
       │   │ └───────────────┘ │
       ▼   └───────────────────┘

### timerCtx

其实Context最牛的功能我觉得还是timerCtx,先来看一下这个功能一个简单的例子。

```go
func main() {
   ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
   defer cancel()

result := make(chan int, 3)

for i := 0; i < 10; i++ {
      go func(i int) {
         for {
            select {
            case <-ctx.Done():
               fmt.Println("return")
               return
            case result <- i:

}
         }
      }(i)
   }

for {
      select {
      case r := <-result:
         fmt.Println(r)
      case <-ctx.Done():
         return
      }
   }
}
```

其实这个ctx就是定义了一个具有超时功能的上下文,一般可以应用在可能会长时间执行的任务上，如果该任务长时间执行，我们可以设置一个ctx,超时时间到来，goroutine就从该任务返回，不会造成任务失控的情况。继续看下timerCtx的结构

```go
// A timerCtx carries a timer and a deadline. It embeds a cancelCtx to
// implement Done and Err. It implements cancel by stopping its timer then
// delegating to cancelCtx.cancel.
// 通过计时器去实现任务的取消
type timerCtx struct {
	//内部也是继承了cancelCtx,所以也具有取消的能力
	cancelCtx
	
	timer *time.Timer // Under cancelCtx.mu.

//超时时间点
	deadline time.Time
}
```

WithTimeout其实内部会调用WithDeadline,我们分析下该方法

```go
func WithDeadline(parent Context, d time.Time) (Context, CancelFunc) {
   if parent == nil {
      panic("cannot create context from nil parent")
   }

//父节点早于子节点指定时间，直接返回父节点，因为后面设置其实没意义
   if cur, ok := parent.Deadline(); ok && cur.Before(d) {
      // The current deadline is already sooner than the new one.
      return WithCancel(parent)
   }

c := &timerCtx{
      cancelCtx: newCancelCtx(parent),
      deadline:  d,
   }
   propagateCancel(parent, c)
   dur := time.Until(d)

//时间已经到了，就直接cancel
   if dur <= 0 {
      c.cancel(true, DeadlineExceeded) // deadline has already passed
      return c, func() { c.cancel(false, Canceled) }
   }
   c.mu.Lock()
   defer c.mu.Unlock()

//用定时器去处理延迟cancel
   if c.err == nil {
      c.timer = time.AfterFunc(dur, func() {
         c.cancel(true, DeadlineExceeded)
      })
   }
   return c, func() { c.cancel(true, Canceled) }
}
```

这里有3点要注意：

1. 如果子Ctx超过了父Ctx则，直接使用父Ctx
2. 如果时间已经到期，则直接Cancel
3. 否则就注册一个定时器在未来一个时间执行

这里比较关心的是，他内部其实维护了一个定时器，就是那么简单而已！！！在分析一下对应的cancel方法

```go
func (c *timerCtx) cancel(removeFromParent bool, err error) {
   c.cancelCtx.cancel(false, err)
   if removeFromParent {
      // Remove this timerCtx from its parent cancelCtx's children.
      removeChild(c.cancelCtx.Context, c)
   }
   c.mu.Lock()

//关闭定时器
   if c.timer != nil {
      c.timer.Stop()
      c.timer = nil
   }
   c.mu.Unlock()
}
```

比cancelCtx多了停止定时器的操作。

### valueCtx

WithValue很容易建立valueCtx,valueCtx结构如下

```go
// A valueCtx carries a key-value pair. It implements Value for that key and
// delegates all other calls to the embedded Context.
type valueCtx struct {
   Context
   key, val interface{}
}

func WithValue(parent Context, key, val interface{}) Context {
	if parent == nil {
		panic("cannot create context from nil parent")
	}
	if key == nil {
		panic("nil key")
	}
	if !reflectlite.TypeOf(key).Comparable() {
		panic("key is not comparable")
	}
	return &valueCtx{parent, key, val}
}
```

WithValue只能通过添加的方式把父Ctx和新Ctx建立连接，很显然整个Ctx看起来应该就是一棵树一样。比如如下的代码

```go
type TrackId string

func main() {
	ctx1 := context.WithValue(context.Background(), TrackId("2021"), "123456")

ctx2 := context.WithValue(ctx1, TrackId("2020"), "111111")
	ctx3 := context.WithValue(ctx1, TrackId("2020"), "222222")

ctx4 := context.WithValue(ctx2, TrackId("2019"), "333333")
	ctx5 := context.WithValue(ctx2, TrackId("2019"), "444444")

ctx6 := context.WithValue(ctx3, TrackId("2018"), "555555")
	ctx7 := context.WithValue(ctx3, TrackId("2018"), "666666")

var wg sync.WaitGroup
	wg.Add(1)

go func() {
		defer wg.Done()

fmt.Println("ctx4 ", ctx4.Value(TrackId("2019")))
		fmt.Println("ctx5 ", ctx5.Value(TrackId("2019")))

fmt.Println("ctx6 ", ctx6.Value(TrackId("2020")))
		fmt.Println("ctx7 ", ctx7.Value(TrackId("2021")))
	}()

wg.Wait()
}
```

其实就是会建立这样一棵树结构

```
                                         ┌────────────────────────┐
                                         │ ┌********************┐ │
                                         │ *     Background     * │
                                         │ └********************┘ │
                                         └────────────┬───────────┘
                                                      │
                                                      │
                                         ┌────────────▼───────────┐
                                         │                        │
                                         │    ctx1(val:123456)    ◀──────Step3
                                         │                        │
                                         └────────────┬───────────┘
                                                      │
                           ┌──────────────────────────┴──────────────────────────┐
                           │                                                     │
                           │                                                     │
              ┌────────────▼───────────┐                            ┌────────────▼───────────┐
              │                        │                            │                        │
              │     ctx2(val:111111)   │                            │    ctx3(val:222222)    ◀──────Step2
              │                        │                            │                        │
              └────────────┬───────────┘                            └────────────┬───────────┘
                           │                                                     │
             ┌─────────────┴────────────┐                          ┌─────────────┴────────────┐
             │                          │                          │                          │
             │                          │                          │                          │
┌────────────▼───────────┐ ┌────────────▼───────────┐ ┌────────────▼───────────┐ ┌────────────▼───────────┐
│                        │ │                        │ │                        │ │                        │
│     ctx4(val:333333)   │ │     ctx5(val:444444)   │ │    ctx6(val:555555)    │ │    ctx7(val:666666)    ◀──────Step1
│                        │ │                        │ │                        │ │                        │
└────────────────────────┘ └────────────────────────┘ └────────────────────────┘ └────────────────────────┘
```

当然如果调用挂载的节点越多，这棵树就越大，而遍历这棵树找value信息就越慢，事实上上找value信息就是通过往上递归遍历的方法来查找的。

```go
func (c *valueCtx) Value(key interface{}) interface{} {
   if c.key == key {
      return c.val
   }
   return c.Context.Value(key)
}
```

比如ctx7.Value(TrackId("2021"))就需要通过Step1,2,3才能找到最终的value:123456。

### 总结

timerCtx,cancelCtx可以认为是管理goroutine生命周期的一类Ctx,另外valueCtx是传递参数作用的Ctx,使用的场景其实有区别,比较会误用的是valueCtx。这里摘取了一些使用中容易挖的坑，其实要使用好它，还真不容易！

beanstalkd源码解析(2)

作者: nbboy
时间: 2020-10-23
分类: 软件架构,软件工程,设计模式,C
评论

###开篇
  在开篇先介绍下beanstalkd的一些基础结构,然后分析下第二个问题。
####基础结构

```c
//最小堆结构(优先队列)
struct Heap {
    int     cap;
    int     len;
    void    **data;
    Less    less;
    Record  rec;
};

//ADT
int   heapinsert(Heap *h, void *x);//向堆中插入一个新元素
void* heapremove(Heap *h, int k);//删除堆中的一个元素
```

```c
//集合结构
struct ms {
    size_t used, cap, last;
    void **items;
    ms_event_fn oninsert, onremove;
};

//ADT
void ms_init(ms a, ms_event_fn oninsert, ms_event_fn onremove);//初始化
void ms_clear(ms a);//清空
int ms_append(ms a, void *item);//追加
int ms_remove(ms a, void *item);//移除
int ms_contains(ms a, void *item);//是否包含
void *ms_take(ms a);//获取一个
```

#### 业务结构

```c
//任务对象,客户端可以调度的一个单元
struct job {
	//持久化字段,这些字段都被写入到WAL日志中
    Jobrec r; // persistent fields; these get written to the wal

/* bookeeping fields; these are in-memory only */
    char pad[6];
	
	//任务属于的管道对象
    tube tube;
	
	//在链表中的上个和下个任务
    job prev, next; /* linked list of jobs */
	
	//在哈希表中的下个任务
    job ht_next; /* Next job in a hash table list */
	
	//任务在当前堆中的位置
    size_t heap_index; /* where is this job in its current heap */
	
    File *file;
    job  fnext;
    job  fprev;
    void *reserver;
    int walresv;
    int walused;

char body[]; // written separately to the wal
};

#define make_job(pri,delay,ttr,body_size,tube) make_job_with_id(pri,delay,ttr,body_size,tube,0)

job allocate_job(int body_size);//分配一个job
job make_job_with_id(uint pri, int64 delay, int64 ttr,
             int body_size, tube tube, uint64 id);//创建job
void job_free(job j);//释放job

/* Lookup a job by job ID */
job job_find(uint64 job_id);//查找

/* the void* parameters are really job pointers */
void job_setheappos(void*, int);
int job_pri_less(void*, void*);
int job_delay_less(void*, void*);

job job_copy(job j);//拷贝job

const char * job_state(job j);//获取job状态

int job_list_any_p(job head);
job job_remove(job j);//移除job
void job_insert(job head, job j);//插入job

/* for unit tests */
size_t get_all_jobs_used(void);
```

```c
//管道对象,用来抽象同一种消息类型
struct tube {
	//被引用次数,用于监控管道被使用的情况
    uint refs;
	
	//管道名称
    char name[MAX_TUBE_NAME_LEN];
	
	//预备执行队列
    Heap ready;
	
	//延迟执行队列
    Heap delay;
	
	
	//等待的连接集合
    struct ms waiting; /* set of conns */
	
	//统计结构,stats系列命令用到该结构
    struct stats stat;
	
	//正被使用的次数
    uint using_ct;
	
	//正被监听的次数
    uint watching_ct;
    int64 pause;
	
	//
    int64 deadline_at;
	
	//
    struct job buried;
};

tube make_tube(const char *name);//创建一个tube
void tube_dref(tube t);
void tube_iref(tube t);
tube tube_find(const char *name);//查找
tube tube_find_or_make(const char *name);
```

####服务有关结构

```c
/*
对服务器对象的封装,全局单例
*/
struct Server {
	//端口
    char *port;
	
	//地址
    char *addr;
	
	//用户
    char *user;

//WAL日志对象
    Wal    wal;
	
	//服务器开启的套接字
    Socket sock;
	
	//客户端连接池
    Heap   conns;
};
void srvserve(Server *srv);
void srvaccept(Server *s, int ev);
```

```c
//客户端链接对象
struct Conn {
	//服务器对象
    Server *srv;
	
	//Client套接字
    Socket sock;
    char   state;
    char   type;
	
	//下一个连接对象
    Conn   *next;
	
	//操作的管道
    tube   use;
	
	//运作时间
    int64  tickat;      // time at which to do more work
	
	//在链接池中的索引
    int    tickpos;     // position in srv->conns
	
	//最近的任务
    job    soonest_job; // memoization of the soonest job
    int    rw;          // currently want: 'r', 'w', or 'h'
	
	//阻塞超时的时间
    int    pending_timeout;
    char   halfclosed;

//命令
    char cmd[LINE_BUF_SIZE]; // this string is NOT NUL-terminated
    int  cmd_len;
    int  cmd_read;

//回复
    char *reply;
    int  reply_len;
    int  reply_sent;
    char reply_buf[LINE_BUF_SIZE]; // this string IS NUL-terminated

// How many bytes of in_job->body have been read so far. If in_job is NULL
    // while in_job_read is nonzero, we are in bit bucket mode and
    // in_job_read's meaning is inverted -- then it counts the bytes that
    // remain to be thrown away.
    int in_job_read;
    job in_job; // a job to be read from the client

job out_job;
    int out_job_sent;

//监听的集合
    struct ms  watch;
	
	//正在被消费的任务列表
    struct job reserved_jobs; // linked list header
};

int  connless(Conn *a, Conn *b);
void connrec(Conn *c, int i);
void connwant(Conn *c, int rw);
void connsched(Conn *c);
void connclose(Conn *c);
void connsetproducer(Conn *c);
void connsetworker(Conn *c);
job  connsoonestjob(Conn *c);
int  conndeadlinesoon(Conn *c);
int conn_ready(Conn *c);
```

需要说明的是,文件有关的结构和WAL有关的结构，就不在这里贴出来了，因为我也没整明白。
###分析
  看第二个问题.回顾下刚才代码

```c
//投递任务到队列
static int
enqueue_job(Server *s, job j, int64 delay, char update_store)
{
    int r;

j->reserver = NULL;
    if (delay) {
		//任务开始执行时间
        j->r.deadline_at = nanoseconds() + delay;
        r = heapinsert(&j->tube->delay, j);
        if (!r) return 0;
		//设置为被等待执行状态
        j->r.state = Delayed;
    } else {//立即执行的任务就投递到预备队列
```

如果是延迟任务，就放入到tube的延迟堆中,且状态是delay

那么什么时候状态变为ready,然后放入ready堆中以供客户端去消费？

客户端投递一个任务后，下一个硬件断点看下值的变化

```c
(gdb) watch all_jobs_init[4].r.
body_size    bury_ct      created_at   deadline_at  delay        id           kick_ct      pri          release_ct   reserve_ct   state        timeout_ct   ttr
(gdb) watch all_jobs_init[4].r.state
Hardware watchpoint 6: all_jobs_init[4].r.state
(gdb) info breakpoints
Num     Type           Disp Enb Address            What
6       hw watchpoint  keep y                      all_jobs_init[4].r.state
```

然后客户端开始消费任务，时间到期后，即命中断点

```c
(gdb) c
Continuing.
Hardware watchpoint 6: all_jobs_init[4].r.state

Old value = 4 '\004'
New value = 1 '\001'
enqueue_job (s=0x611520 <srv>, j=0x62a970, delay=0, update_store=0 '\000') at prot.c:479
479	        ready_ct++;
(gdb) bt
#0  enqueue_job (s=0x611520 <srv>, j=0x62a970, delay=0, update_store=0 '\000') at prot.c:479
#1  0x0000000000409e90 in prottick (s=0x611520 <srv>) at prot.c:1904
#2  0x000000000040b2ee in srvserve (s=0x611520 <srv>) at serv.c:51
#3  0x000000000040d1d4 in main (argc=1, argv=0x7fffffffe488) at main.c:100
```

从堆栈中看出，是在prottick中进行调度。截取关键代码片段

```c
while ((j = delay_q_peek())) {
        d = j->r.deadline_at - now;
        if (d > 0) {
            period = min(period, d);
            break;
        }
        j = delay_q_take();
		//从delay里取出最近需要执行的job,放入ready队列里
        r = enqueue_job(s, j, 0, 0);
        if (r < 1) bury_job(s, j, 0); /* out of memory, so bury it */
    }
```

从上面代码可以看出把delay堆中的任务往ready堆中移动的过程,不过这段代码是怎样触发的?

往上翻一下就可以找到答案

```c
for (;;) {
        period = prottick(s);

//sockXXX都是对平台的网络抽象
        int rw = socknext(&sock, period);
        if (rw == -1) {
            twarnx("socknext");
            exit(1);
        }

//调用上面的接收链接回调函数
        if (rw) {
			//accept开始接受网络请求
            sock->f(sock->x, rw);
        }
    }

int
socknext(Socket **s, int64 timeout)
{
    int r;
    struct epoll_event ev;

r = epoll_wait(epfd, &ev, 1, (int)(timeout/1000000));
  ...
```

触发的条件有两个要么timeout到期，要么有新事件到来,不明白epoll事件机制的可以复习下epoll,那如何

计算这个超时时间呢？

for (i = 0; i < tubes.used; i++) {
        t = tubes.items[i];
        d = t->deadline_at - now;
        if (t->pause && d <= 0) {
            t->pause = 0;
            process_queue();
        }
        else if (d > 0) {
            period = min(period, d);
        }
    }

while (s->conns.len) {
        Conn *c = s->conns.data[0];
        d = c->tickat - now;
        if (d > 0) {
            period = min(period, d);
            break;
        }

heapremove(&s->conns, 0);
        conn_timeout(c);
    }
```

也就是根据job,tube,conn这几个对象计算一个最小时间戳

总结下第二个问题:

1.通过epoll_wait(其他平台也类似)来等待超时或者网络事件,而超时时候就做定时逻辑，包括移动delay堆中的job到ready堆中

这篇贴的代码有点多，下一篇再说第三个问题…

beanstalkd源码解析(1)

作者: nbboy
时间: 2020-10-23
分类: 软件架构,软件工程,设计模式,C
评论

###简介
  Beanstalkd是一个简单、高效的工作队列系统，其最初设计目的是通过后台异步执行耗时任务方式降低高容量Web应用的页面延时。而其简单、轻量、易用等特点，和对任务优先级、延时 超时重发等控制，以及众多语言版本的客户端的良好支持，使其可以很好的在各种需要队列系统的场景中应用。
  其采用的是生产者、消费者模式,借鉴了memcached的设计,协议也很简单。先了解下其基础概念：
#####job - 任务
  job是一个需要异步处理的任务，是Beanstalkd中的基本单元，job需要放在一个tube中。Beanstalkd中的任务（job）类似于其它队列系统中的消息（message）的概念。
#####tube - 管道
  管道即某一种类型的任务队列，其类似与消息的主题（topic）,是Producer和Consumer的操作对象。一个Beanstalkd中可以有多个管道, 每个管道都有自己的发布者（Producer）和消费者Consumer，管道之间互相不影响。
#####producer - 生产者
  任务（job）的生产者，通过put命令来将一个job放到一个tube中。
#####consumer - 消费者
  任务（job）的消费者，通过reserve、release、bury、delete命令来获取或改变job的状态。
  在网络上找了一副图，描述的是其状态流转情况：
![](https://chenxf.org/usr/uploads/2020/10/2669523929.png)
一个Beanstalkd任务可能会包含以下状态：
**READY** - 需要立即处理的任务。当producer直接put一个任务时，任务就处于READY状态，以等待consumer来处理。当延时 (DELAYED) 任务到期后会自动成为当前READY状态的任务

**DELAYED** - 延迟执行的任务。当任务被延时put时，任务就处于DELAYED状态。等待时间过后，任务会被迁移到READY状态。当消费者处理任务后，可以用将消息再次放回DELAYED队列延迟执行

**RESERVED** - 已经被消费者获取，正在执行的任务。当consumer获取了当前READY的任务后，该任务的状态就会迁移到RESERVED状态，这时其它的consumer就不能再操作该任务。Beanstalkd会检查任务是否在TTR(time-to-run)内完成

**BURIED** - 保留的任务，这时任务不会被执行，也不会消失。当consumer完成该任务后，可以选择delete、release或者bury操作。
delete后，任务会被删除，生命周期结束；release操作可以重新把任务状态迁移回READY状态或DELAYED状态，使其他consumer可以继续获取和执行该任务。bury会拔任务休眠，等需要该任务时，再将休眠的任务kick回READY；也可能过delete删除BURIED状态的任务

**DELETED** - 消息被删除，Beanstalkd不再维持这些消息。即任务生命周期结束。
###分析
  请原谅我罗里吧嗦得说了那么多，现在直接进入主题。我一般看源码都是带着问题去看，这样效率会高很多，这次也不例外，先提出几个问题吧。
- **beanstalkd如何维护job的状态？**
- **delay状态的job怎么修改为ready？**
- **如何实现优先级？**
- **数据是如何持久化?**

对于问题1可以关注use,put,reserve,delete等命令的处理逻辑

######use 是生产者命令
(lldb) bt

* thread #1, queue = 'com.apple.main-thread', stop reason = step over
  * frame #0: 0x000000010000c011 beanstalkd`make_and_insert_tube(name="abc") at tube.c:70
    frame #1: 0x000000010000bfd9 beanstalkd`tube_find_or_make(name="abc") at tube.c:96
    frame #2: 0x00000001000087d0 beanstalkd`dispatch_cmd(c=0x0000000100400430) at prot.c:1554
    frame #3: 0x0000000100006ca5 beanstalkd`do_cmd(c=0x0000000100400430) at prot.c:1686
    frame #4: 0x00000001000067ca beanstalkd`conn_data(c=0x0000000100400430) at prot.c:1726
    frame #5: 0x0000000100006623 beanstalkd`h_conn(fd=6, which=114, c=0x0000000100400430) at prot.c:1868
    frame #6: 0x0000000100005968 beanstalkd`prothandle(c=0x0000000100400430, ev=114) at prot.c:1880
    frame #7: 0x000000010000bb75 beanstalkd`srvserve(s=0x00000001000104b8) at serv.c:63
    frame #8: 0x000000010000e19f beanstalkd`main(argc=1, argv=0x00007ffeefbff370) at main.c:100
    frame #9: 0x00007fff6df26085 libdyld.dylib`start + 1

######跟踪put命令

* * (lldb) bt
    * thread #1, queue = 'com.apple.main-thread', stop reason = breakpoint 6.1
      * frame #0: 0x0000000100004f99 beanstalkd`enqueue_job(s=0x00000001000104b8, j=0x0000000100203b90, delay=0, update_store='\x01') at prot.c:466
        frame #1: 0x00000001000070bf beanstalkd`enqueue_incoming_job(c=0x0000000100203850) at prot.c:867
        frame #2: 0x0000000100006dfb beanstalkd`maybe_enqueue_incoming_job(c=0x0000000100203850) at prot.c:1160
        frame #3: 0x000000010000757f beanstalkd`dispatch_cmd(c=0x0000000100203850) at prot.c:1279
        frame #4: 0x0000000100006ca5 beanstalkd`do_cmd(c=0x0000000100203850) at prot.c:1686
        frame #5: 0x00000001000067ca beanstalkd`conn_data(c=0x0000000100203850) at prot.c:1726
        frame #6: 0x0000000100006623 beanstalkd`h_conn(fd=6, which=114, c=0x0000000100203850) at prot.c:1868
        frame #7: 0x0000000100005968 beanstalkd`prothandle(c=0x0000000100203850, ev=114) at prot.c:1880
        frame #8: 0x000000010000bb75 beanstalkd`srvserve(s=0x00000001000104b8) at serv.c:63
        frame #9: 0x000000010000e19f beanstalkd`main(argc=1, argv=0x00007ffeefbff370) at main.c:100
        frame #10: 0x00007fff6df26085 libdyld.dylib`start + 1

put命令主要是在ready堆中插入job

```c
frame #0: 0x0000000100004f99 beanstalkd`enqueue_job(s=0x00000001000104b8, j=0x0000000100500370, delay=0, update_store='\x01') at prot.c:466
   463 	{
   464 	    int r;
   465
-> 466 	    j->reserver = NULL;
   467 	    if (delay) {
   468 			//任务开始执行时间
   469 	        j->r.deadline_at = nanoseconds() + delay;
Target 0: (beanstalkd) stopped.
(lldb) l
   470 	        r = heapinsert(&j->tube->delay, j);
   471 	        if (!r) return 0;
   472 			//设置为被等待执行状态
   473 	        j->r.state = Delayed;
   474 	    } else {//立即执行的任务就投递到预备队列
   475 	        r = heapinsert(&j->tube->ready, j);
   476 	        if (!r) return 0;
(lldb) l
   477 			//设置状态为预备执行状态
   478 	        j->r.state = Ready;
   479 	        ready_ct++;
   480
   481 			//对于紧急任务,进行额外的跟踪处理
   482 	        if (j->r.pri < URGENT_THRESHOLD) {
   483 	            global_stat.urgent_ct++;
(lldb) l
   484 	            j->tube->stat.urgent_ct++;
   485 	        }
   486 	    }
```

######reserve命令

(lldb) bt

* thread #1, queue = 'com.apple.main-thread', stop reason = step in
  * frame #0: 0x000000010000ac0c beanstalkd`enqueue_waiting_conn(c=0x0000000100500420) at prot.c:687
    frame #1: 0x00000001000097f0 beanstalkd`wait_for_job(c=0x0000000100500420, timeout=3600) at prot.c:1024
    frame #2: 0x00000001000079cd beanstalkd`dispatch_cmd(c=0x0000000100500420) at prot.c:1358
    frame #3: 0x0000000100006ca5 beanstalkd`do_cmd(c=0x0000000100500420) at prot.c:1686
    frame #4: 0x00000001000067ca beanstalkd`conn_data(c=0x0000000100500420) at prot.c:1726
    frame #5: 0x0000000100006623 beanstalkd`h_conn(fd=6, which=114, c=0x0000000100500420) at prot.c:1868
    frame #6: 0x0000000100005968 beanstalkd`prothandle(c=0x0000000100500420, ev=114) at prot.c:1880
    frame #7: 0x000000010000bb75 beanstalkd`srvserve(s=0x00000001000104b8) at serv.c:63
    frame #8: 0x000000010000e19f beanstalkd`main(argc=1, argv=0x00007ffeefbff370) at main.c:100
    frame #9: 0x00007fff6df26085 libdyld.dylib`start + 1

```c
1344	    case OP_RESERVE: /* FALLTHROUGH */
   1345	        /* don't allow trailing garbage */
-> 1346	        if (type == OP_RESERVE && c->cmd_len != CMD_RESERVE_LEN + 2) {
   1347	            return reply_msg(c, MSG_BAD_FORMAT);
   1348	        }
   1349
Target 0: (beanstalkd) stopped.
(lldb) l
   1350	        op_ct[type]++;
   1351	        connsetworker(c);
   1352
   1353	        if (conndeadlinesoon(c) && !conn_ready(c)) {
   1354	            return reply_msg(c, MSG_DEADLINE_SOON);
   1355	        }
   1356
(lldb) l
   1357	        /* try to get a new job for this guy */
   1358	        wait_for_job(c, timeout);
   1359	        process_queue();
   1360	        break;
```

把当前conn先加入到waitting集合中

```c
* thread #1, queue = 'com.apple.main-thread', stop reason = step over
    frame #0: 0x00000001000097e7 beanstalkd`wait_for_job(c=0x0000000100500420, timeout=3600) at prot.c:1024
   1021	wait_for_job(Conn *c, int timeout)
   1022	{
   1023	    c->state = STATE_WAIT;
-> 1024	    enqueue_waiting_conn(c);
   1025
   1026	    /* Set the pending timeout to the requested timeout amount */
   1027	    c->pending_timeout = timeout;
Target 0: (beanstalkd) stopped.
(lldb) s
Process 40697 stopped
* thread #1, queue = 'com.apple.main-thread', stop reason = step in
    frame #0: 0x000000010000ac0c beanstalkd`enqueue_waiting_conn(c=0x0000000100500420) at prot.c:687
   684 	    tube t;
   685 	    size_t i;
   686
-> 687 	    global_stat.waiting_ct++;
   688 	    c->type |= CONN_TYPE_WAITING;
   689 	    for (i = 0; i < c->watch.used; i++) {
   690 	        t = c->watch.items[i];
Target 0: (beanstalkd) stopped.
(lldb) l
   691 	        t->stat.waiting_ct++;
     //加入到tube的等待集合中
   692 	        ms_append(&t->waiting, c);
   693 	    }
   694 	}
```

状态从ready到reserve

```c
thread #1, queue = 'com.apple.main-thread', stop reason = step in
    frame #0: 0x0000000100006030 beanstalkd`reserve_job(c=0x0000000100500420, j=0x0000000100203b90) at prot.c:380
   377 	static void
   378 	reserve_job(Conn *c, job j)
   379 	{
-> 380 	    j->r.deadline_at = nanoseconds() + j->r.ttr;
   381 	    global_stat.reserved_ct++; /* stats */
   382 	    j->tube->stat.reserved_ct++;
   383 	    j->r.reserve_ct++;
Target 0: (beanstalkd) stopped.
(lldb) l
  //改变状态为reserved
   384 	    j->r.state = Reserved;
      //插入到conn的reserved_jobs堆中
   385 	    job_insert(&c->reserved_jobs, j);
   386 	    j->reserver = c;
   387 	    c->pending_timeout = -1;
   388 	    if (c->soonest_job && j->r.deadline_at < c->soonest_job->r.deadline_at) {
   389 	        c->soonest_job = j;
   390 	    }
(lldb) l
   391 	    return reply_job(c, j, MSG_RESERVED);
   392 	}
```

######简单看下delete操作

```c
case OP_DELETE:
-> 1363	        errno = 0;
   1364	        id = strtoull(c->cmd + CMD_DELETE_LEN, &end_buf, 10);
   1365	        if (errno) return reply_msg(c, MSG_BAD_FORMAT);
   1366	        op_ct[type]++;
Target 0: (beanstalkd) stopped.
(lldb) l
   1367
  //查找到job后，从几个堆中进行移除操作
   1368	        j = job_find(id);
   1369	        j = remove_reserved_job(c, j) ? :
   1370	            remove_ready_job(j) ? :
   1371	            remove_buried_job(j) ? :
   1372	            remove_delayed_job(j);
   1373
(lldb) l
   1374	        if (!j) return reply(c, MSG_NOTFOUND, MSG_NOTFOUND_LEN, STATE_SENDWORD);
   1375
   1376	        j->tube->stat.total_delete_ct++;
   1377
   1378	        j->r.state = Invalid;
   1379	        r = walwrite(&c->srv->wal, j);
   1380	        walmaint(&c->srv->wal);
(lldb) l
  //最后进行释放操作
   1381	        job_free(j);
```

######对第一个问题做下简单总结:

1.ready,delay堆都在tube对象进行维护，而reserved堆则在conn对象进行维护。

2.消费客户端连接都先放到对应watch的tube等待集合中(tube.waiting)，在process_queue函数中进行统一消费，其用伪代码如下描述,这段是比较关键的部分

```python
while j = next_eligible_job()://从优先堆中取出job
	job_remove(tube.ready, j)//ready状态堆中移除job
	job_insert(conn.reserved, j)//把job插入客户端连接的reserved集合中
```

下一篇再说第二个问题...