Error Handling in Go(注解)

• 作者: nbboy
• 时间: 2020-11-27
• 分类: 软件工程,Golang
• 评论

原链接:https://medium.com/gett-engineering/error-handling-in-go-53b8a7112d04 ###begin   Go’s approach to error handling is one of its most controversial and misused features. In this article, you’ll learn how Go approaches errors and understand how they work under the hood. You’ll explore a couple of different approaches to it, take a look at the go source code and the standard library for some insights about how errors work and how to work with them. You’ll learn how Type Assertions play an important role in handling them, and see upcoming changes to error handling, planned to be released in Go 2. **Introduction**   First thing’s first: Errors in Go are not Exceptions. Dave Cheney wrote an epic blog post about it, so I’ll refer you to it and summarise: In other languages, you are uncertain if a function may throw an exception or not. Instead of throwing exceptions, Go functions support multiple return values, and by convention, this ability is commonly used to return the function’s result along with an error variable. 注解:go error不是异常，和其他语言不一样，因为其他不确定在运行时会抛出什么异常，而golang是通过返回值来做的    If your function can fail for some reason, you should probably return the predeclared error type from it. By convention, returning an error signals the caller there was a problem, and returning nil represents no error. This way, you’re letting the caller understand there was a problem, and let them deal with it: whoever calls your function knows they should not rely on the result before checking the error. If error is not nil, it is their responsibility to check for it and handle it (log, return, serve, invoke some retry/cleanup mechanism, etc.). 注解:调用者需要为被调用者的error进行判断和负责,可以进行处理或者直接再次返回给上层处理    These snippets are very common in Go, and some see them as a whole lot of boiler plate code. The compiler treats unused variables as compilation errors, so if you’re not going to check the error, you should assign them to the blank identifier. But as convenient as it is, errors should not be ignored. 注解：在golang里，这段代码非常通用，对于错误可以用"_"来忽略，或者处理，还是需要尽量处理error    Returning the error along with the results, along with Go’s strict type system, makes programming errors that much harder to write. You should always assume the function’s value is corrupted unless you’ve checked the error it returned, and by assigning the error to the blank identifier, you explicitly ignore that your function’s value may be corrupt.   Go does have a panic and recover mechanism, which is also described in another detailed Go blog post. But these are not meant to mimic exceptions. In the words of Dave, “When you panic in Go, you’re freaking out, it’s not someone else’s problem, it’s game over man”. They’re fatal, and lead to a crash in your program. Rob Pike coined the “Don’t Panic” proverb, which is self-explanatory: you should probably avoid it, and return errors instead. - “Errors are values.” - “Don’t just check errors, handle them gracefully” - “Don’t panic” - all of Rob Pike’s Go Proverbs 注解：你看大神也说了，error就是一个普通值，而且不要用panic(需要自己决定) ###Under the hood ####The error Interface   Under the hood, the error type is a simple single method interface, and if you’re not familiar with it, I highly recommend going over this post in the official Go Blog. 注解:可以看到error是一个接口，并且非常简单，只有一个Error方法,我们只需要实现它，就实现了error    It’s easy to implement your own errors, and there are various approaches to custom structs implementing the Error() string method. Any struct implementing this one method is considered a valid error value, and can be returned as such. Let’s explore a few such approaches. ####The built-in errorString struct   The most commonly used and widespread implementation of the error interface is the built-in errorString struct. It is the leanest implementation you can think of. 注解：通常的实现就是errorString struct    You can see its simplistic implementation here. All it does is hold a string, and that string is returned by the Error method. That error string can be formatted with data by us, say, with fmt.Sprintf. But other than that, it does not pack any other capabilities. If you use the built-in errors.New or fmt.Errorf, you are already using it. 注解：能看到非常简单的实现，内部只包括一个string,也能用errors.New 或者 fmt.Errorf来创建这个error  github.com/pkg/errors   Another simple example is the pkg/errors package. Not to be confused with the built-in errors package you learned about earlier, this package provides additional important capabilities such as error wrapping, unwrapping, formatting and stack trace recording. You can install the package by running go get github.com/pkg/errors . 注解：这个包不会和原生的error对象冲突，就是说兼容的，而且实现了wrapping, unwrapping, 格式化 和 错误栈记录功能    In those cases where you need to attach stack traces to your errors, or attach necessary debugging information to the error, using this package’s New or Errorf functions provides errors that already record your stack trace, and you can attach simple metadata using its formatting capabilities. Errorf implements the fmt.Formatter interface, meaning you can format it using the fmt package runes ( %s , %v , %+v etc). 注解：这个包的功能都其实非常有用，因为我们有时候就是需要一些错误的上下文信息，而且他也能用更好的格式化形式去提供    This package also introduces the errors.Wrap and errors.Wrapf functions. These functions add context to an error with a message and stack trace at the point they were called. This way, instead of simply returning an error, you can wrap it with its context and important debug data. 注解：他的errors.Wrap ， errors.Wrapf能把上下文的一些信息和堆栈添加到错误的原信息里去，为后面的排错提供很重要的调试依据。    Errors wrapping errors support the Cause() error method, that returns their inner error. Also, they can be used with the errors.Cause(err error) error function which retrieves the underlying inner-most error within an error. 注解：他的errors.Cause(err error)方法通过回溯的方法能够找到错误最原始的现场。

Defer, Panic, and Recover(注解)

• 作者: nbboy
• 时间: 2020-11-27
• 分类: 软件工程,Golang
• 评论

原链接:https://blog.golang.org/defer-panic-and-recover Andrew Gerrand 4 August 2010 Go has the usual mechanisms for control flow: if, for, switch, goto. It also has the go statement to run code in a separate goroutine. Here I'd like to discuss some of the less common ones: defer, panic, and recover. A defer statement pushes a function call onto a list. The list of saved calls is executed after the surrounding function returns. Defer is commonly used to simplify functions that perform various clean-up actions. For example, let's look at a function that opens two files and copies the contents of one file to the other: ```go func CopyFile(dstName, srcName string) (written int64, err error) { src, err := os.Open(srcName) if err != nil { return } dst, err := os.Create(dstName) if err != nil { return } written, err = io.Copy(dst, src) dst.Close() src.Close() return } ``` 注解:如果没有defer的情况下，代码是以上写的一样，有时候忘记close资源，会有资源泄露问题 This works, but there is a bug. If the call to os.Create fails, the function will return without closing the source file. This can be easily remedied by putting a call to src.Close before the second return statement, but if the function were more complex the problem might not be so easily noticed and resolved. By introducing defer statements we can ensure that the files are always closed: ```go func CopyFile(dstName, srcName string) (written int64, err error) { src, err := os.Open(srcName) if err != nil { return } defer src.Close() dst, err := os.Create(dstName) if err != nil { return } defer dst.Close() return io.Copy(dst, src) } ``` 注解:引入defer后，代码干净了很多 Defer statements allow us to think about closing each file right after opening it, guaranteeing that, regardless of the number of return statements in the function, the files will be closed. The behavior of defer statements is straightforward and predictable. There are three simple rules: A deferred function's arguments are evaluated when the defer statement is evaluated. In this example, the expression "i" is evaluated when the Println call is deferred. The deferred call will print "0" after the function returns. ```go func a() { i := 0 defer fmt.Println(i) i++ return } ``` 注解：这里打印的是0，而非1，也就是说在defer的时候是立即计算的 Deferred function calls are executed in Last In First Out order after the surrounding function returns. This function prints "3210": ```go func b() { for i := 0; i < 4; i++ { defer fmt.Print(i) } } ``` 注解：观察结果，可以看到defer的顺序是类似FIFO结果，也就是先出现的最后执行 Deferred functions may read and assign to the returning function's named return values. In this example, a deferred function increments the return value i after the surrounding function returns. Thus, this function returns 2: ```go func c() (i int) { defer func() { i++ }() return 1 } ``` 注解：defer也可以做到，在返回之前修改返回值 This is convenient for modifying the error return value of a function; we will see an example of this shortly. Panic is a built-in function that stops the ordinary flow of control and begins panicking. When the function F calls panic, execution of F stops, any deferred functions in F are executed normally, and then F returns to its caller. To the caller, F then behaves like a call to panic. The process continues up the stack until all functions in the current goroutine have returned, at which point the program crashes. Panics can be initiated by invoking panic directly. They can also be caused by runtime errors, such as out-of-bounds array accesses. Recover is a built-in function that regains control of a panicking goroutine. Recover is only useful inside deferred functions. During normal execution, a call to recover will return nil and have no other effect. If the current goroutine is panicking, a call to recover will capture the value given to panic and resume normal execution. Here's an example program that demonstrates the mechanics of panic and defer: ```go package main import "fmt" func main() { f() fmt.Println("Returned normally from f.") } func f() { defer func() { if r := recover(); r != nil { fmt.Println("Recovered in f", r) } }() fmt.Println("Calling g.") g(0) fmt.Println("Returned normally from g.") } func g(i int) { if i > 3 { fmt.Println("Panicking!") panic(fmt.Sprintf("%v", i)) } defer fmt.Println("Defer in g", i) fmt.Println("Printing in g", i) g(i + 1) } ``` 注解：在同一个goroutine里，用recover在defer里可以捕获到panic掉的异常 The function g takes the int i, and panics if i is greater than 3, or else it calls itself with the argument i+1. The function f defers a function that calls recover and prints the recovered value (if it is non-nil). Try to picture what the output of this program might be before reading on. The program will output: Calling g. Printing in g 0 Printing in g 1 Printing in g 2 Printing in g 3 Panicking! Defer in g 3 Defer in g 2 Defer in g 1 Defer in g 0 Recovered in f 4 Returned normally from f. If we remove the deferred function from f the panic is not recovered and reaches the top of the goroutine's call stack, terminating the program. This modified program will output: Calling g. Printing in g 0 Printing in g 1 Printing in g 2 Printing in g 3 Panicking! Defer in g 3 Defer in g 2 Defer in g 1 Defer in g 0 panic: 4 panic PC=0x2a9cd8 [stack trace omitted] For a real-world example of panic and recover, see the json package from the Go standard library. It encodes an interface with a set of recursive functions. If an error occurs when traversing the value, panic is called to unwind the stack to the top-level function call, which recovers from the panic and returns an appropriate error value (see the 'error' and 'marshal' methods of the encodeState type in encode.go). The convention in the Go libraries is that even when a package uses panic internally, its external API still presents explicit error return values. Other uses of defer (beyond the file.Close example given earlier) include releasing a mutex: ```go mu.Lock() defer mu.Unlock() printing a footer: ``` ```go printHeader() defer printFooter() and more. ``` 注解:在锁场景也是defer用的最多的场景 In summary, the defer statement (with or without panic and recover) provides an unusual and powerful mechanism for control flow. It can be used to model a number of features implemented by special-purpose structures in other programming languages. Try it out. 总结: - defer在golang中确实很爽的特性,主要用在资源cleanup的时候，比如锁，或者文件 - 需要小心的是在defer闭包中，传入参数的值是立即被计算的，也就是定义那时候的值 - 另外一个典型的应用场景就是在defer中进行recover被panic的问题

Error Handling in Go that Every Beginner should Know(注解)

• 作者: nbboy
• 时间: 2020-11-27
• 分类: Golang
• 评论

原链接:https://hussachai.medium.com/error-handling-in-go-a-quick-opinionated-guide-9199dd7c7f76 Go doesn’t have exceptions, so it doesn’t have try…catch or anything similar. How can we handle errors in Go then? There are two common methods for handling errors in Go — Multiple return values and panic. Error Handling using Multiple Return Values We can take advantage of multiple return values feature by adding an error struct to returned values. By convention, an error value will be on the right and a result value will be on the left. This convention kind of bugs me because it doesn’t sound right that the error is on the right side :D. ```go func Marshal(v interface{}) ([]byte, error) { e := &encodeState{} err := e.marshal(v, encOpts{escapeHTML: true}) if err != nil { return nil, err } return e.Bytes(), nil } ``` 注:Golang里一般采用多值返回，按照惯例，错误也会在最右边返回! The above code snippet is taken from the json package in the standard library. Notice that the error is a type. It is an interface declared in the builtin package. ```go type error interface { Error() string } ``` The problem of this error handling pattern is that there is no enforcement from a compiler. It is up to you on what and how your function returns an error. You can put an error struct in any position of the returned values. It can be on the left or it can be in the middle if you have more than two returned values. Actually, it doesn’t have to be an error type at all. It’s all up to you. However, you may want to follow the standard convention if you don’t want your colleagues to come to your desk and ask you to change (forcefully). Note that the error interface contains one function which is Error() string. There is no short way to create an anonymous type in Go. You have to define a struct and use method sets to associate a function to your struct. // Define your Error struct ```go type MyError struct { msg string } ``` // Create a function Error() string and associate it to the struct. ```go func (error *MyError) Error() string { return error.msg } ``` // Now you can construct an error object using MyError struct. ```go func ThisFunctionReturnError() error { return &MyError{"custom error"} } ``` 注:我们能实现error接口来扩展自己的类型，不过这种方式最大的问题是没有出错的上下文堆栈。 This code looks very tedious. Okay, you can create a common package and share this piece of code from there. Fortunately, the standard library’s got you covered. There is the errorString struct defined in the errors package. Stack Traces We can refactor our previous code to use the built-in error struct as follows. import "errors" ```go func ThisFunctionReturnError() error { return errors.New("custom error") } ``` It now looks much better, doesn’t it? You may still have a question about stack traces based on the error interface you have seen . Unfortunately, the standard errors does not come with stack traces. Without stack traces, it is very difficult to debug when you have no idea where the error occurs. The error could pass through 10 function calls before it gets to the code that prints it out. Many people have this exact same problem and they created awesome projects to handle this issue. palantir/stacktrace, go-erros/errors, and pkg/errors are one of them. In the future, Go standard library may come with the richer error type and it could render all of those libraries obsolete. But, for now, pick one of them because you definitely need it. I personally like pkg/errors the most since it’s compatible with the standard package. Basically, you can just change the import statement from “errors” to “github.com/pkg/errors” and call it a day. ```go import ( "github.com/pkg/errors" "fmt" ) func A() error { return errors.New("NullPointerException") } func B() error { return A() } func main() { fmt.Printf("Error: %+v", B()) } ``` 注:正是由于上面的问题，所以社区出现了一些包可以做到捕捉错误上下文的功能，作者最推荐的是pkg/errors。 If you want to print stack traces instead of a plain error message, you have to use %+v instead of %v in the format pattern, and the stack traces will look similar as below. This library has many other functions that are very useful, but be careful that those functions are not defined in the standard package. You’re creating a library lock-in when you use other functions besides errors.New. Error: NullPointerException main.A /Users/hussachai/Go/src/awesomeProject/error_handling.go:9 main.B /Users/hussachai/Go/src/awesomeProject/error_handling.go:13 main.main /Users/hussachai/Go/src/awesomeProject/error_handling.go:17 runtime.main /usr/local/opt/go/libexec/src/runtime/proc.go:198 runtime.goexit /usr/local/opt/go/libexec/src/runtime/asm_amd64.s:2361 The Ugly Error Checking When you use multiple return values to handle errors, eventually you may end up with a deeply nested if-block which is one of the pain point of this error handling style. ```go func EatOrange() (*C, error) { var err error var a string a, err = GetA() if err == nil { var b string b, err = GetB(a) if err == nil { var c string c, err = GetC(b) if err == nil { return c, nil } } } return nil, err } ``` 注：不推荐这种动动波式代码，应该出现错误就返回，马上判断掉，比如如下代码形式。 ```go cli := client.NewClient(handler.Tcp, "localhost:1234") caller, err := cli.Dial() if err != nil { log.Fatalln("client dial error") } var result Result err = caller.Call("Calc.Add", &Req{Num1: 5, Num2: 6}, &result) if err != nil { log.Fatalln("cli call error") } ``` The levels of nesting can go terribly deeper than the example above. Composing a decision tree using the if-control-flow is pretty ugly IMO, yet it is surely easy to understand. This is a trade off that you have to balance it out while you’re crafting your code. Opting Out an Error In Go, you can omit one or more variable assignments by putting an underscore on the position of the variable you want to omit on the left-hand-side. The underscore in this context is called blank identifier. Since an error is a plain struct, you can apply the blank identifier on them as other returned values. ```go result, _ := EatOrange() // omit the error assignment _, err := EatOrange() // omit the result struct ``` Defer, Panic, and Recover Here you go. The one resembling an exception that you’re familiar with. I mean, you can break the flow by throwing it (panic). Go has a pretty unique way to handle the panicking (exception). Go’s blog covers this topic very well (of course). If you are feeling lazy and want to know how it works quickly, please read on. Defer is a language mechanism that puts your function call into a stack. The deferred functions will be executed in reverse order when the host function finishes regardless of whether a panic is called or not. The defer mechanism is very useful for cleaning up resources. ```go resp, _ := http.Get("http://golang.org") defer func () { if resp != nil { resp.Body.Close() } }() body, _ := ioutil.ReadAll(resp.Body) ``` Panic is a built-in function that stops the normal execution flow. The deferred functions are still run as usual. ```go func panic(v interface{}) ``` See how it works in action: https://play.golang.org/p/sfkGfBo04d3 When you call panic and you don’t handle it, the execution flow stops, all deferred functions are executed in reverse order, and stack traces are printed at the end. You can pass any types into the panic function. However, I’d recommend passing an error struct because you will not lose stack traces when you recover a panic. Of course, you have to use one of those errors libraries I mentioned earlier. Recover is a built-in function that returns the value passing from a panic call. This function must be called in a deferred function. Otherwise, it always returns nil. package main ```go import ( "fmt" "github.com/pkg/errors" ) func A() { defer fmt.Println("A") defer func() { if r := recover(); r != nil { fmt.Printf("Panic: %+v\n", r) } }() B() } func B() { defer fmt.Println("B") C() } func C() { defer fmt.Println("C") Break() } func Break() { defer fmt.Println("D") panic(errors.New("the show must go on")) } func main() { A() } ``` 注:golang另外一种错误处理方式是采用panic,当发送严重错误的时候可以抛出panic。 Converting Panicking into a Returned Error. Sometimes, you don’t want to stop the whole execution flow due to a panic, but you want to report an error back to a caller as a returned value. In this case, you have to recover a panicking goroutine and grab an error struct obtaining from the recover function, and then pass it to a variable. It’s not difficult as it sounds, and that is how named returned values come into play. ```go func Perform() (err error) { defer func() { if r := recover(); r != nil { err = r.(error) } }() GoesWrong() return } func GoesWrong() { panic(errors.New("Fail")) } func main() { err := Perform() fmt.Println(err) } ``` 注:recover能够捕获panic抛出的异常,不过必须是在defer闭包里,这时候就可以有时机把panic转换成error返回了,所以在panic的时候，往往会把error带进去。 The deferred functions will be executed after a function call, but before a return statement. So, you have an opportunity to set a returned variable before a return statement gets executed. I hope this article is useful to you. Please let me know if you have any comments or suggestions. Thank you! 总结: - 可以用第三方库区包裹标准库的error,这样可以让最外层获得错误的stacktrace - 限制用panic,如果不小心panic，必须把它给recover,然后转换成error,友好得给前端,不让服务hang

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,那如何 计算这个超时时间呢？ ```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 */ } 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的状态。   在网络上找了一副图，描述的是其状态流转情况：  一个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集合中 ``` 下一篇再说第二个问题...

1. 1
2. 2
3. 3
4. 4
5. ...
6. 7
7. 后一页 »