{"action":"created","comment":{"url":"https://api.github.com/repos/jbenet/go-ipfs/pulls/comments/23232541","id":23232541,"diff_hunk":"@@ -0,0 +1,169 @@\n+// TODO FIXME name me\n+package worker\n+\n+import (\n+\t\"container/list\"\n+\t\"errors\"\n+\t\"time\"\n+\n+\tcontext \"github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context\"\n+\tprocess \"github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess\"\n+\tblocks \"github.com/jbenet/go-ipfs/blocks\"\n+\texchange \"github.com/jbenet/go-ipfs/exchange\"\n+\tutil \"github.com/jbenet/go-ipfs/util\"\n+)\n+\n+var log = util.Logger(\"blockservice\")\n+\n+var DefaultConfig = Config{\n+\tNumWorkers:       1,\n+\tClientBufferSize: 0,\n+\tWorkerBufferSize: 0,\n+}\n+\n+type Config struct {\n+\t// NumWorkers sets the number of background workers that provide blocks to\n+\t// the exchange.\n+\tNumWorkers int\n+\n+\t// ClientBufferSize allows clients of HasBlock to send up to\n+\t// |ClientBufferSize| blocks without blocking.\n+\tClientBufferSize int\n+\n+\t// WorkerBufferSize can be used in conjunction with NumWorkers to reduce\n+\t// communication-coordination within the worker.\n+\tWorkerBufferSize int\n+}\n+\n+// TODO FIXME name me\n+type Worker struct {\n+\t// added accepts blocks from client\n+\tadded    chan *blocks.Block\n+\texchange exchange.Interface\n+\n+\t// workQueue is owned by the client worker\n+\t// process manages life-cycle\n+\tprocess process.Process\n+}\n+\n+func NewWorker(e exchange.Interface, c Config) *Worker {\n+\tif c.NumWorkers < 1 {\n+\t\tc.NumWorkers = 1 // provide a sane default\n+\t}\n+\tw := &Worker{\n+\t\texchange: e,\n+\t\tadded:    make(chan *blocks.Block, c.ClientBufferSize),\n+\t\tprocess:  process.WithParent(process.Background()), // internal management\n+\t}\n+\tw.start(c)\n+\treturn w\n+}\n+\n+func (w *Worker) HasBlock(b *blocks.Block) error {\n+\tselect {\n+\tcase <-w.process.Closed():\n+\t\treturn errors.New(\"blockservice worker is closed\")\n+\tcase w.added <- b:\n+\t\treturn nil\n+\t}\n+}\n+\n+func (w *Worker) Close() error {\n+\tlog.Debug(\"blockservice provide worker is shutting down...\")\n+\treturn w.process.Close()\n+}\n+\n+func (w *Worker) start(c Config) {\n+\n+\tworkerChan := make(chan *blocks.Block, c.WorkerBufferSize)\n+\n+\t// clientWorker handles incoming blocks from |w.added| and sends to\n+\t// |workerChan|. This will never block the client.\n+\tw.process.Go(func(proc process.Process) {\n+\t\tdefer close(workerChan)\n+\n+\t\tvar workQueue BlockList\n+\t\tfor {\n+\n+\t\t\t// take advantage of the fact that sending on nil channel always\n+\t\t\t// blocks so that a message is only sent if a block exists\n+\t\t\tsendToWorker := workerChan\n+\t\t\tnextBlock := workQueue.Pop()\n+\t\t\tif nextBlock == nil {\n+\t\t\t\tsendToWorker = nil\n+\t\t\t}\n+\n+\t\t\tselect {\n+\n+\t\t\t// if worker is ready and there's a block to process, send the\n+\t\t\t// block\n+\t\t\tcase sendToWorker <- nextBlock:\n+\t\t\tcase <-time.Tick(5 * time.Second):\n+\t\t\t\tif workQueue.Len() > 0 {\n+\t\t\t\t\tlog.Debugf(\"%d blocks in blockservice provide queue...\", workQueue.Len())\n+\t\t\t\t}\n+\t\t\tcase block := <-w.added:\n+\t\t\t\tif nextBlock != nil {\n+\t\t\t\t\tworkQueue.Push(nextBlock) // missed the chance to send it\n+\t\t\t\t}\n+\t\t\t\t// if the client sends another block, add it to the queue.\n+\t\t\t\tworkQueue.Push(block)\n+\t\t\tcase <-proc.Closing():\n+\t\t\t\treturn\n+\t\t\t}\n+\t\t}\n+\t})\n+\n+\tfor i := 0; i < c.NumWorkers; i++ {\n+\t\t// reads from |workerChan| until process closes\n+\t\tw.process.Go(func(proc process.Process) {\n+\t\t\tctx, cancel := context.WithCancel(context.Background())\n+\n+\t\t\t// shuts down an in-progress HasBlock operation\n+\t\t\tproc.Go(func(proc process.Process) {\n+\t\t\t\t<-proc.Closing()\n+\t\t\t\tcancel()\n+\t\t\t})\n+\n+\t\t\tfor {\n+\t\t\t\tselect {\n+\t\t\t\tcase <-proc.Closing():\n+\t\t\t\t\treturn\n+\t\t\t\tcase block, ok := <-workerChan:\n+\t\t\t\t\tif !ok {\n+\t\t\t\t\t\treturn\n+\t\t\t\t\t}\n+\t\t\t\t\tif err := w.exchange.HasBlock(ctx, block); err != nil {\n+\t\t\t\t\t\t// TODO log event?","path":"blockservice/worker/worker.go","position":137,"original_position":137,"commit_id":"2ec5622b51b893b776bb0a23beb724d1178e77b9","original_commit_id":"2ec5622b51b893b776bb0a23beb724d1178e77b9","user":{"login":"jbenet","id":138401,"avatar_url":"https://avatars.githubusercontent.com/u/138401?v=3","gravatar_id":"","url":"https://api.github.com/users/jbenet","html_url":"https://github.com/jbenet","followers_url":"https://api.github.com/users/jbenet/followers","following_url":"https://api.github.com/users/jbenet/following{/other_user}","gists_url":"https://api.github.com/users/jbenet/gists{/gist_id}","starred_url":"https://api.github.com/users/jbenet/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jbenet/subscriptions","organizations_url":"https://api.github.com/users/jbenet/orgs","repos_url":"https://api.github.com/users/jbenet/repos","events_url":"https://api.github.com/users/jbenet/events{/privacy}","received_events_url":"https://api.github.com/users/jbenet/received_events","type":"User","site_admin":false},"body":"yeah, log the error, at least to debug. since we propagate errors up, this could be something worth looking at","created_at":"2015-01-20T16:11:44Z","updated_at":"2015-01-20T16:11:44Z","html_url":"https://github.com/jbenet/go-ipfs/pull/606#discussion_r23232541","pull_request_url":"https://api.github.com/repos/jbenet/go-ipfs/pulls/606","_links":{"self":{"href":"https://api.github.com/repos/jbenet/go-ipfs/pulls/comments/23232541"},"html":{"href":"https://github.com/jbenet/go-ipfs/pull/606#discussion_r23232541"},"pull_request":{"href":"https://api.github.com/repos/jbenet/go-ipfs/pulls/606"}}},"pull_request":{"url":"https://api.github.com/repos/jbenet/go-ipfs/pulls/606","id":27683210,"html_url":"https://github.com/jbenet/go-ipfs/pull/606","diff_url":"https://github.com/jbenet/go-ipfs/pull/606.diff","patch_url":"https://github.com/jbenet/go-ipfs/pull/606.patch","issue_url":"https://api.github.com/repos/jbenet/go-ipfs/issues/606","number":606,"state":"open","locked":false,"title":"fix #424: when adding blocks, provide to exchange asynchronously","user":{"login":"briantigerchow","id":734339,"avatar_url":"https://avatars.githubusercontent.com/u/734339?v=3","gravatar_id":"","url":"https://api.github.com/users/briantigerchow","html_url":"https://github.com/briantigerchow","followers_url":"https://api.github.com/users/briantigerchow/followers","following_url":"https://api.github.com/users/briantigerchow/following{/other_user}","gists_url":"https://api.github.com/users/briantigerchow/gists{/gist_id}","starred_url":"https://api.github.com/users/briantigerchow/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/briantigerchow/subscriptions","organizations_url":"https://api.github.com/users/briantigerchow/orgs","repos_url":"https://api.github.com/users/briantigerchow/repos","events_url":"https://api.github.com/users/briantigerchow/events{/privacy}","received_events_url":"https://api.github.com/users/briantigerchow/received_events","type":"User","site_admin":false},"body":"@whyrusleeping @jbenet RFCR\r\n\r\n* :bulb: In debug mode, the worker periodically prints the number of blocks in the provide queue.\r\n* :beetle: When a file is added twice in quick succession, blocks are queued up to be provided twice. We'll probably want to de-dupe.","created_at":"2015-01-20T14:56:36Z","updated_at":"2015-01-20T16:11:44Z","closed_at":null,"merged_at":null,"merge_commit_sha":"534a66e6c7ea2f863224b73f768436a184420787","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/jbenet/go-ipfs/pulls/606/commits","review_comments_url":"https://api.github.com/repos/jbenet/go-ipfs/pulls/606/comments","review_comment_url":"https://api.github.com/repos/jbenet/go-ipfs/pulls/comments/{number}","comments_url":"https://api.github.com/repos/jbenet/go-ipfs/issues/606/comments","statuses_url":"https://api.github.com/repos/jbenet/go-ipfs/statuses/2ec5622b51b893b776bb0a23beb724d1178e77b9","head":{"label":"jbenet:fix-424-blockservice-async-HasBlock","ref":"fix-424-blockservice-async-HasBlock","sha":"2ec5622b51b893b776bb0a23beb724d1178e77b9","user":{"login":"jbenet","id":138401,"avatar_url":"https://avatars.githubusercontent.com/u/138401?v=3","gravatar_id":"","url":"https://api.github.com/users/jbenet","html_url":"https://github.com/jbenet","followers_url":"https://api.github.com/users/jbenet/followers","following_url":"https://api.github.com/users/jbenet/following{/other_user}","gists_url":"https://api.github.com/users/jbenet/gists{/gist_id}","starred_url":"https://api.github.com/users/jbenet/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jbenet/subscriptions","organizations_url":"https://api.github.com/users/jbenet/orgs","repos_url":"https://api.github.com/users/jbenet/repos","events_url":"https://api.github.com/users/jbenet/events{/privacy}","received_events_url":"https://api.github.com/users/jbenet/received_events","type":"User","site_admin":false},"repo":{"id":21233191,"name":"go-ipfs","full_name":"jbenet/go-ipfs","owner":{"login":"jbenet","id":138401,"avatar_url":"https://avatars.githubusercontent.com/u/138401?v=3","gravatar_id":"","url":"https://api.github.com/users/jbenet","html_url":"https://github.com/jbenet","followers_url":"https://api.github.com/users/jbenet/followers","following_url":"https://api.github.com/users/jbenet/following{/other_user}","gists_url":"https://api.github.com/users/jbenet/gists{/gist_id}","starred_url":"https://api.github.com/users/jbenet/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jbenet/subscriptions","organizations_url":"https://api.github.com/users/jbenet/orgs","repos_url":"https://api.github.com/users/jbenet/repos","events_url":"https://api.github.com/users/jbenet/events{/privacy}","received_events_url":"https://api.github.com/users/jbenet/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/jbenet/go-ipfs","description":"ipfs implementation in go","fork":false,"url":"https://api.github.com/repos/jbenet/go-ipfs","forks_url":"https://api.github.com/repos/jbenet/go-ipfs/forks","keys_url":"https://api.github.com/repos/jbenet/go-ipfs/keys{/key_id}","collaborators_url":"https://api.github.com/repos/jbenet/go-ipfs/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/jbenet/go-ipfs/teams","hooks_url":"https://api.github.com/repos/jbenet/go-ipfs/hooks","issue_events_url":"https://api.github.com/repos/jbenet/go-ipfs/issues/events{/number}","events_url":"https://api.github.com/repos/jbenet/go-ipfs/events","assignees_url":"https://api.github.com/repos/jbenet/go-ipfs/assignees{/user}","branches_url":"https://api.github.com/repos/jbenet/go-ipfs/branches{/branch}","tags_url":"https://api.github.com/repos/jbenet/go-ipfs/tags","blobs_url":"https://api.github.com/repos/jbenet/go-ipfs/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/jbenet/go-ipfs/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/jbenet/go-ipfs/git/refs{/sha}","trees_url":"https://api.github.com/repos/jbenet/go-ipfs/git/trees{/sha}","statuses_url":"https://api.github.com/repos/jbenet/go-ipfs/statuses/{sha}","languages_url":"https://api.github.com/repos/jbenet/go-ipfs/languages","stargazers_url":"https://api.github.com/repos/jbenet/go-ipfs/stargazers","contributors_url":"https://api.github.com/repos/jbenet/go-ipfs/contributors","subscribers_url":"https://api.github.com/repos/jbenet/go-ipfs/subscribers","subscription_url":"https://api.github.com/repos/jbenet/go-ipfs/subscription","commits_url":"https://api.github.com/repos/jbenet/go-ipfs/commits{/sha}","git_commits_url":"https://api.github.com/repos/jbenet/go-ipfs/git/commits{/sha}","comments_url":"https://api.github.com/repos/jbenet/go-ipfs/comments{/number}","issue_comment_url":"https://api.github.com/repos/jbenet/go-ipfs/issues/comments/{number}","contents_url":"https://api.github.com/repos/jbenet/go-ipfs/contents/{+path}","compare_url":"https://api.github.com/repos/jbenet/go-ipfs/compare/{base}...{head}","merges_url":"https://api.github.com/repos/jbenet/go-ipfs/merges","archive_url":"https://api.github.com/repos/jbenet/go-ipfs/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/jbenet/go-ipfs/downloads","issues_url":"https://api.github.com/repos/jbenet/go-ipfs/issues{/number}","pulls_url":"https://api.github.com/repos/jbenet/go-ipfs/pulls{/number}","milestones_url":"https://api.github.com/repos/jbenet/go-ipfs/milestones{/number}","notifications_url":"https://api.github.com/repos/jbenet/go-ipfs/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/jbenet/go-ipfs/labels{/name}","releases_url":"https://api.github.com/repos/jbenet/go-ipfs/releases{/id}","created_at":"2014-06-26T08:14:34Z","updated_at":"2015-01-20T14:24:47Z","pushed_at":"2015-01-20T14:41:36Z","git_url":"git://github.com/jbenet/go-ipfs.git","ssh_url":"git@github.com:jbenet/go-ipfs.git","clone_url":"https://github.com/jbenet/go-ipfs.git","svn_url":"https://github.com/jbenet/go-ipfs","homepage":"https://github.com/jbenet/ipfs","size":22880,"stargazers_count":251,"watchers_count":251,"language":"Go","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":34,"mirror_url":null,"open_issues_count":125,"forks":34,"open_issues":125,"watchers":251,"default_branch":"master"}},"base":{"label":"jbenet:master","ref":"master","sha":"5c4be75146b1297fef64007ce3733521e60c772a","user":{"login":"jbenet","id":138401,"avatar_url":"https://avatars.githubusercontent.com/u/138401?v=3","gravatar_id":"","url":"https://api.github.com/users/jbenet","html_url":"https://github.com/jbenet","followers_url":"https://api.github.com/users/jbenet/followers","following_url":"https://api.github.com/users/jbenet/following{/other_user}","gists_url":"https://api.github.com/users/jbenet/gists{/gist_id}","starred_url":"https://api.github.com/users/jbenet/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jbenet/subscriptions","organizations_url":"https://api.github.com/users/jbenet/orgs","repos_url":"https://api.github.com/users/jbenet/repos","events_url":"https://api.github.com/users/jbenet/events{/privacy}","received_events_url":"https://api.github.com/users/jbenet/received_events","type":"User","site_admin":false},"repo":{"id":21233191,"name":"go-ipfs","full_name":"jbenet/go-ipfs","owner":{"login":"jbenet","id":138401,"avatar_url":"https://avatars.githubusercontent.com/u/138401?v=3","gravatar_id":"","url":"https://api.github.com/users/jbenet","html_url":"https://github.com/jbenet","followers_url":"https://api.github.com/users/jbenet/followers","following_url":"https://api.github.com/users/jbenet/following{/other_user}","gists_url":"https://api.github.com/users/jbenet/gists{/gist_id}","starred_url":"https://api.github.com/users/jbenet/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jbenet/subscriptions","organizations_url":"https://api.github.com/users/jbenet/orgs","repos_url":"https://api.github.com/users/jbenet/repos","events_url":"https://api.github.com/users/jbenet/events{/privacy}","received_events_url":"https://api.github.com/users/jbenet/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/jbenet/go-ipfs","description":"ipfs implementation in go","fork":false,"url":"https://api.github.com/repos/jbenet/go-ipfs","forks_url":"https://api.github.com/repos/jbenet/go-ipfs/forks","keys_url":"https://api.github.com/repos/jbenet/go-ipfs/keys{/key_id}","collaborators_url":"https://api.github.com/repos/jbenet/go-ipfs/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/jbenet/go-ipfs/teams","hooks_url":"https://api.github.com/repos/jbenet/go-ipfs/hooks","issue_events_url":"https://api.github.com/repos/jbenet/go-ipfs/issues/events{/number}","events_url":"https://api.github.com/repos/jbenet/go-ipfs/events","assignees_url":"https://api.github.com/repos/jbenet/go-ipfs/assignees{/user}","branches_url":"https://api.github.com/repos/jbenet/go-ipfs/branches{/branch}","tags_url":"https://api.github.com/repos/jbenet/go-ipfs/tags","blobs_url":"https://api.github.com/repos/jbenet/go-ipfs/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/jbenet/go-ipfs/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/jbenet/go-ipfs/git/refs{/sha}","trees_url":"https://api.github.com/repos/jbenet/go-ipfs/git/trees{/sha}","statuses_url":"https://api.github.com/repos/jbenet/go-ipfs/statuses/{sha}","languages_url":"https://api.github.com/repos/jbenet/go-ipfs/languages","stargazers_url":"https://api.github.com/repos/jbenet/go-ipfs/stargazers","contributors_url":"https://api.github.com/repos/jbenet/go-ipfs/contributors","subscribers_url":"https://api.github.com/repos/jbenet/go-ipfs/subscribers","subscription_url":"https://api.github.com/repos/jbenet/go-ipfs/subscription","commits_url":"https://api.github.com/repos/jbenet/go-ipfs/commits{/sha}","git_commits_url":"https://api.github.com/repos/jbenet/go-ipfs/git/commits{/sha}","comments_url":"https://api.github.com/repos/jbenet/go-ipfs/comments{/number}","issue_comment_url":"https://api.github.com/repos/jbenet/go-ipfs/issues/comments/{number}","contents_url":"https://api.github.com/repos/jbenet/go-ipfs/contents/{+path}","compare_url":"https://api.github.com/repos/jbenet/go-ipfs/compare/{base}...{head}","merges_url":"https://api.github.com/repos/jbenet/go-ipfs/merges","archive_url":"https://api.github.com/repos/jbenet/go-ipfs/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/jbenet/go-ipfs/downloads","issues_url":"https://api.github.com/repos/jbenet/go-ipfs/issues{/number}","pulls_url":"https://api.github.com/repos/jbenet/go-ipfs/pulls{/number}","milestones_url":"https://api.github.com/repos/jbenet/go-ipfs/milestones{/number}","notifications_url":"https://api.github.com/repos/jbenet/go-ipfs/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/jbenet/go-ipfs/labels{/name}","releases_url":"https://api.github.com/repos/jbenet/go-ipfs/releases{/id}","created_at":"2014-06-26T08:14:34Z","updated_at":"2015-01-20T14:24:47Z","pushed_at":"2015-01-20T14:41:36Z","git_url":"git://github.com/jbenet/go-ipfs.git","ssh_url":"git@github.com:jbenet/go-ipfs.git","clone_url":"https://github.com/jbenet/go-ipfs.git","svn_url":"https://github.com/jbenet/go-ipfs","homepage":"https://github.com/jbenet/ipfs","size":22880,"stargazers_count":251,"watchers_count":251,"language":"Go","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":34,"mirror_url":null,"open_issues_count":125,"forks":34,"open_issues":125,"watchers":251,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/jbenet/go-ipfs/pulls/606"},"html":{"href":"https://github.com/jbenet/go-ipfs/pull/606"},"issue":{"href":"https://api.github.com/repos/jbenet/go-ipfs/issues/606"},"comments":{"href":"https://api.github.com/repos/jbenet/go-ipfs/issues/606/comments"},"review_comments":{"href":"https://api.github.com/repos/jbenet/go-ipfs/pulls/606/comments"},"review_comment":{"href":"https://api.github.com/repos/jbenet/go-ipfs/pulls/comments/{number}"},"commits":{"href":"https://api.github.com/repos/jbenet/go-ipfs/pulls/606/commits"},"statuses":{"href":"https://api.github.com/repos/jbenet/go-ipfs/statuses/2ec5622b51b893b776bb0a23beb724d1178e77b9"}}}}