forked from appscode/g2
-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Enable worker to reconnect to gearman indefinitely (#7)
* Add agent reference to ErrorHandler args to allow specific connection recovery * Chang disconnect logic to attempt reconnecting to server indefinitely * Change Agent struct back to un-exported "agent" * Change Agent member back to un-exported "agent" * Removed outdated comment * Rename agent.Addr to agent.addr since don't need it to be exported * Add/update persistent reconnect behavior to client, worker * Add persistent initial connection to worker, consolidate Connect() and Reconnect() * Add persistent initial connection and re-connect behavior to client * Replace NewConnected client constructor with a universal version which takes connection open/close handlers as arguments. * Add io error handling in a few io places where it was missing * Update readme with updated function name * Add a sense of ownership of which thread (read or write) is reconnecting Problem: suppose readLoop() encountered an error on the connection, it starts the reconnect process, but suppose at nearly the same time the writeLoop() thread was using the same connection handle and also encountered an error (since connection is broken). This fix is to prevent both of them attempting to reconnect and have one of them claim the ownership of reconnect process and have the other thread merely wait for it to complete and exit reconnect, and then go on with using the new conn handle via getConn(). * Fix handle naming to camel case * Remove orphaned comment * Implement erroring out of orphaned waiters on chan expected by using channel close Also changed couple of error events in agent and worker to cause reconnection. * Some logging for investigation * Make sure worker purges orphaned tasks when associated gearman dies * Allow Write to bypass the agent lock As result the previous Write() function becams identical to unexported write() so we export it. * Rearrange agent locking and Grab() placement. Make sure Grab is called after each successful execution. NOTE: that it's also called when reconnections are made. * Remove appscode/go/log, and remove got handle and got result messages * Address PR comments. * Revert Split to SplitN etc in response.go * Introduce a writeReconnectCleanup function with a compact write/reconnect/cleanup op * Add a todo for a more sophisticated gearman connection health checker * Remove commented out code * Change usages of log in worker and client to new logHandler, make it optional to pass Also rename some handles to match a common pattern such as [sS]omethingHandler * Fix writeLoop exit on reconnect * Work around using the Lock in client, optimize read and write loop reconnect behavior * Streamline locking and new job grab/requesting Add current jobs count for diagnostics * Make improvements based on PR feedback. * move channels back into client * restore NewConnected constructor * add safe error casting for recover output * atomic loads for conn * Update on PR comments, load atomics * Move channel close to go after conn close * Add conn comparison in read loop * Make sure write loop checks connection before series of writes for a single transmission This is to help the integrity of the writes in case reconnects occur mid-transmission * Change some names to camelCase * Close outbound and expected after they are replaced, make access atomic. * Move channel close back up before reconnecting Reason is we want any hung up submit, echo, status callers to error out before we attempt reconnecting because it might be minutes before the gearman is back up and we don't want them hanging.
- Loading branch information
Showing
14 changed files
with
715 additions
and
249 deletions.
There are no files selected for viewing
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,124 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "log" | ||
| "time" | ||
|
|
||
| "github.com/quantcast/g2/client" | ||
| rt "github.com/quantcast/g2/pkg/runtime" | ||
| ) | ||
|
|
||
| func logHandler(level client.LogLevel, message ...string) { | ||
| switch level { | ||
| case client.Error: | ||
| log.Println("Error:", message) | ||
| case client.Warning: | ||
| log.Println("Warning", message) | ||
| case client.Info: | ||
| log.Println("Info:", message) | ||
| case client.Debug: | ||
| log.Println("Debug", message) | ||
| } | ||
| } | ||
|
|
||
| func main() { | ||
| // Set the autoinc id generator | ||
| // You can write your own id generator | ||
| // by implementing IdGenerator interface. | ||
| // client.IdGen = client.NewAutoIncId() | ||
|
|
||
| logs.InitLogs() | ||
| logs.FlushLogs() | ||
| c, err := client.NewNetClient(rt.Network, "127.0.0.1:4730") | ||
| if err != nil { | ||
| log.Fatalln(err) | ||
| } | ||
| defer c.Close() | ||
| c.ErrorHandler = func(e error) { | ||
| log.Println("ErrorHandler Received:", e) | ||
| } | ||
| echo := []byte("Hello\x00 world") | ||
| echomsg, err := c.Echo(echo) | ||
| if err != nil { | ||
| log.Printf("Error in Echo:", err) | ||
| } else { | ||
| log.Println("EchoMsg:", string(echomsg)) | ||
| } | ||
|
|
||
| print_result := true | ||
| print_update := false | ||
| print_status := false | ||
|
|
||
| jobHandler := func(resp *client.Response) { | ||
| switch resp.DataType { | ||
| case rt.PT_WorkException: | ||
| fallthrough | ||
| case rt.PT_WorkFail: | ||
| fallthrough | ||
| case rt.PT_WorkComplete: | ||
| if print_result { | ||
| if data, err := resp.Result(); err == nil { | ||
| log.Printf("RESULT: %v, string:%v\n", data, string(data)) | ||
| } else { | ||
| log.Printf("RESULT: %s\n", err) | ||
| } | ||
| } | ||
| case rt.PT_WorkWarning: | ||
| fallthrough | ||
| case rt.PT_WorkData: | ||
| if print_update { | ||
| if data, err := resp.Update(); err == nil { | ||
| log.Printf("UPDATE: %v\n", data) | ||
| } else { | ||
| log.Printf("UPDATE: %v, %s\n", data, err) | ||
| } | ||
| } | ||
| case rt.PT_WorkStatus: | ||
| if print_status { | ||
| if data, err := resp.Status(); err == nil { | ||
| log.Printf("STATUS: %v\n", data) | ||
| } else { | ||
| log.Printf("STATUS: %s\n", err) | ||
| } | ||
| } | ||
| default: | ||
| log.Printf("UNKNOWN: %v", resp.Data) | ||
| } | ||
| } | ||
|
|
||
| log.Println("Press Ctrl-C to exit ...") | ||
|
|
||
| for i := 0; ; i++ { | ||
|
|
||
| if !c.IsConnectionSet() { | ||
| log.Printf("No active connection to server.. waiting...") | ||
| time.Sleep(5 * time.Second) | ||
| continue | ||
| } | ||
|
|
||
| funcName := "ToUpper" | ||
| log.Println("Calling function", funcName, "with data:", echo) | ||
| handle, err := c.Do(funcName, echo, rt.JobNormal, jobHandler) | ||
| if err != nil { | ||
| log.Printf("Do %v ERROR:", funcName, err) | ||
| } | ||
|
|
||
| log.Printf("Calling Status for handle %v", handle) | ||
| status, err := c.Status(handle) | ||
| if err != nil { | ||
| log.Printf("Status: %v, ERROR: %v", status, err) | ||
| } | ||
|
|
||
| funcName = "Foobar" | ||
| log.Println("Calling function", funcName, "with data:", echo) | ||
| _, err = c.Do(funcName, echo, rt.JobNormal, jobHandler) | ||
| if err != nil { | ||
| log.Printf("Do %v ERROR:", funcName, err) | ||
| } | ||
| var sleep_seconds int = 0 | ||
| log.Printf("Finished Cycle %v, sleeping %v seconds", i, sleep_seconds) | ||
| time.Sleep(time.Duration(sleep_seconds) * time.Second) | ||
|
|
||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.