{"action":"created","comment":{"url":"https://api.github.com/repos/m3db/m3msg/pulls/comments/181207623","pull_request_review_id":111775130,"id":181207623,"diff_hunk":"@@ -0,0 +1,258 @@\n+// Copyright (c) 2018 Uber Technologies, Inc.\n+//\n+// Permission is hereby granted, free of charge, to any person obtaining a copy\n+// of this software and associated documentation files (the \"Software\"), to deal\n+// in the Software without restriction, including without limitation the rights\n+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\n+// copies of the Software, and to permit persons to whom the Software is\n+// furnished to do so, subject to the following conditions:\n+//\n+// The above copyright notice and this permission notice shall be included in\n+// all copies or substantial portions of the Software.\n+//\n+// THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\n+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\n+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\n+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\n+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN\n+// THE SOFTWARE.\n+\n+package writer\n+\n+import (\n+\t\"sync\"\n+\n+\t\"github.com/m3db/m3cluster/placement\"\n+\t\"github.com/m3db/m3msg/producer\"\n+\t\"github.com/m3db/m3x/log\"\n+\n+\t\"go.uber.org/atomic\"\n+)\n+\n+type shardWriter interface {\n+\t// Write writes the reference counted data, this needs to be thread safe.\n+\tWrite(d producer.RefCountedData)\n+\n+\t// UpdateInstances updates the instances responsible for this shard.\n+\tUpdateInstances(\n+\t\tinstances []placement.Instance,\n+\t\tcws map[string]consumerWriter,\n+\t)\n+\n+\t// Close closes the writer. It should block until\n+\t// all the data for the given shard have been consumed.\n+\tClose()\n+}\n+\n+type sharedShardWriter struct {\n+\tinstances map[string]struct{}\n+\tmw        messageWriter\n+\tisClosed  *atomic.Bool\n+}\n+\n+func newSharedShardWriter(\n+\tshard uint32,\n+\trouter ackRouter,\n+\tmPool messagePool,\n+\topts Options,\n+) shardWriter {\n+\treplicatedShardID := uint64(shard)\n+\tmw := newMessageWriter(replicatedShardID, mPool, opts)\n+\tmw.Init()\n+\trouter.Register(replicatedShardID, mw)\n+\treturn &sharedShardWriter{\n+\t\tinstances: make(map[string]struct{}),\n+\t\tmw:        mw,\n+\t\tisClosed:  atomic.NewBool(false),\n+\t}\n+}\n+\n+func (w *sharedShardWriter) Write(d producer.RefCountedData) {\n+\tw.mw.Write(d)\n+}\n+\n+// This is not thread safe, must be called in one thread.\n+func (w *sharedShardWriter) UpdateInstances(\n+\tinstances []placement.Instance,\n+\tcws map[string]consumerWriter,\n+) {\n+\tvar (\n+\t\tnewInstancesMap = make(map[string]struct{}, len(instances))\n+\t\ttoBeDeleted     = w.instances\n+\t)\n+\tfor _, instance := range instances {\n+\t\tid := instance.Endpoint()\n+\t\tnewInstancesMap[id] = struct{}{}\n+\t\tif _, ok := toBeDeleted[id]; ok {\n+\t\t\t// Existing instance.\n+\t\t\tdelete(toBeDeleted, id)\n+\t\t\tcontinue\n+\t\t}\n+\t\t// Add the consumer writer to the message writer.\n+\t\tw.mw.AddConsumerWriter(id, cws[id])\n+\t}\n+\tfor id := range toBeDeleted {\n+\t\tw.mw.RemoveConsumerWriter(id)\n+\t}\n+\tw.instances = newInstancesMap\n+}\n+\n+func (w *sharedShardWriter) Close() {\n+\tif !w.isClosed.CAS(false, true) {\n+\t\treturn\n+\t}\n+\tw.mw.Close()\n+}\n+\n+type replicatedShardWriter struct {\n+\tsync.RWMutex\n+\n+\tshard          uint32\n+\tnumberOfShards uint32\n+\tmPool          messagePool\n+\topts           Options\n+\tlogger         log.Logger\n+\n+\tackRouter      ackRouter\n+\treplicaID      uint32\n+\tmessageWriters map[string]messageWriter\n+\tisClosed       bool\n+}\n+\n+func newReplicatedShardWriter(\n+\tshard, numberOfShards uint32,\n+\trouter ackRouter,\n+\tmPool messagePool,\n+\topts Options,\n+) shardWriter {\n+\treturn &replicatedShardWriter{\n+\t\tshard:          shard,\n+\t\tnumberOfShards: numberOfShards,\n+\t\tmPool:          mPool,\n+\t\topts:           opts,\n+\t\tlogger:         opts.InstrumentOptions().Logger(),\n+\t\tackRouter:      router,\n+\t\treplicaID:      0,\n+\t\tmessageWriters: make(map[string]messageWriter),\n+\t\tisClosed:       false,\n+\t}\n+}\n+\n+func (w *replicatedShardWriter) Write(d producer.RefCountedData) {\n+\tw.RLock()\n+\tmws := w.messageWriters\n+\tw.RUnlock()\n+\tfor _, mw := range mws {\n+\t\tmw.Write(d)\n+\t}\n+}\n+\n+// This is not thread safe, must be called in one thread.\n+func (w *replicatedShardWriter) UpdateInstances(\n+\tinstances []placement.Instance,\n+\tcws map[string]consumerWriter,\n+) {\n+\t// TODO: Schedule time after shardcutoff to clean up message writers that\n+\t// are already cutoff. Otherwise it will wait until next placement change\n+\t// to clean up.\n+\tvar (\n+\t\tnewMessageWriters = make(map[string]messageWriter, len(instances))\n+\t\ttoBeClosed        []messageWriter\n+\t\ttoBeAdded         = make(map[placement.Instance]consumerWriter, len(instances))\n+\t\toldMessageWriters = w.messageWriters\n+\t)\n+\n+\tfor _, instance := range instances {\n+\t\tid := instance.Endpoint()\n+\t\tif mw, ok := oldMessageWriters[id]; ok {\n+\t\t\tnewMessageWriters[id] = mw\n+\t\t\t// Existing instance, try to update cutover cutoff times.\n+\t\t\tw.updateCutoverCutoffNanos(mw, instance)\n+\t\t\tcontinue\n+\t\t}\n+\t\t// This is a new instance.\n+\t\ttoBeAdded[instance] = cws[id]\n+\t}\n+\tfor id, mw := range oldMessageWriters {\n+\t\tif _, ok := newMessageWriters[id]; ok {\n+\t\t\t// Still in the new placement.\n+\t\t\tcontinue\n+\t\t}\n+\t\tif instance, cw, ok := anyKeyValueInMap(toBeAdded); ok {","path":"producer/writer/shard_writer.go","position":182,"original_position":182,"commit_id":"f23549c5fe6da89f0edbd2235d4916c39e35c1b6","original_commit_id":"f23549c5fe6da89f0edbd2235d4916c39e35c1b6","user":{"login":"xichen2020","id":12239421,"avatar_url":"https://avatars1.githubusercontent.com/u/12239421?v=4","gravatar_id":"","url":"https://api.github.com/users/xichen2020","html_url":"https://github.com/xichen2020","followers_url":"https://api.github.com/users/xichen2020/followers","following_url":"https://api.github.com/users/xichen2020/following{/other_user}","gists_url":"https://api.github.com/users/xichen2020/gists{/gist_id}","starred_url":"https://api.github.com/users/xichen2020/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/xichen2020/subscriptions","organizations_url":"https://api.github.com/users/xichen2020/orgs","repos_url":"https://api.github.com/users/xichen2020/repos","events_url":"https://api.github.com/users/xichen2020/events{/privacy}","received_events_url":"https://api.github.com/users/xichen2020/received_events","type":"User","site_admin":false},"body":"Okay can you add some comments? It's not clear why this is necessary by just looking at the code.","created_at":"2018-04-12T20:13:06Z","updated_at":"2018-04-12T20:13:06Z","html_url":"https://github.com/m3db/m3msg/pull/8#discussion_r181207623","pull_request_url":"https://api.github.com/repos/m3db/m3msg/pulls/8","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/m3db/m3msg/pulls/comments/181207623"},"html":{"href":"https://github.com/m3db/m3msg/pull/8#discussion_r181207623"},"pull_request":{"href":"https://api.github.com/repos/m3db/m3msg/pulls/8"}},"in_reply_to_id":181183172},"pull_request":{"url":"https://api.github.com/repos/m3db/m3msg/pulls/8","id":181022024,"html_url":"https://github.com/m3db/m3msg/pull/8","diff_url":"https://github.com/m3db/m3msg/pull/8.diff","patch_url":"https://github.com/m3db/m3msg/pull/8.patch","issue_url":"https://api.github.com/repos/m3db/m3msg/issues/8","number":8,"state":"open","locked":false,"title":"Add consumer service writer","user":{"login":"cw9","id":6710599,"avatar_url":"https://avatars0.githubusercontent.com/u/6710599?v=4","gravatar_id":"","url":"https://api.github.com/users/cw9","html_url":"https://github.com/cw9","followers_url":"https://api.github.com/users/cw9/followers","following_url":"https://api.github.com/users/cw9/following{/other_user}","gists_url":"https://api.github.com/users/cw9/gists{/gist_id}","starred_url":"https://api.github.com/users/cw9/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cw9/subscriptions","organizations_url":"https://api.github.com/users/cw9/orgs","repos_url":"https://api.github.com/users/cw9/repos","events_url":"https://api.github.com/users/cw9/events{/privacy}","received_events_url":"https://api.github.com/users/cw9/received_events","type":"User","site_admin":false},"body":"ConsumerServiceWriter maintains a list of shard writers and watches placement updates for the consumer service.\r\n\r\nThere are 2 kinds of ShardWriters, `Shared` and `Replicated`. \r\n- Shared shard writer keeps one message writer for the shard and add/remove consumer writers to the message writer.\r\n- Replicated shard writer keeps one message writer for each consumer instance that owns the shard. Message writers for the same shard ID will be differentiated by replicated shard ID, for example in a placement with 3 replicas and a total of 1000 shards, for shard 123, the replicated shard ID will be 123, 1123 and 2123. Each message writer will have only one consumer writer behind it.\r\n\r\nMessage writer maintains a queue for all the writes for this replicated shard, it writes and retries through the consumer writers behind it.\r\n\r\n@xichen2020 @jeromefroe @jskelcy @m-sandusky ","created_at":"2018-04-11T20:41:28Z","updated_at":"2018-04-12T20:13:06Z","closed_at":null,"merged_at":null,"merge_commit_sha":"829d0419bc42dff2eb8750303424611786e8f52d","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/m3db/m3msg/pulls/8/commits","review_comments_url":"https://api.github.com/repos/m3db/m3msg/pulls/8/comments","review_comment_url":"https://api.github.com/repos/m3db/m3msg/pulls/comments{/number}","comments_url":"https://api.github.com/repos/m3db/m3msg/issues/8/comments","statuses_url":"https://api.github.com/repos/m3db/m3msg/statuses/f23549c5fe6da89f0edbd2235d4916c39e35c1b6","head":{"label":"m3db:chao-writer2","ref":"chao-writer2","sha":"f23549c5fe6da89f0edbd2235d4916c39e35c1b6","user":{"login":"m3db","id":19937171,"avatar_url":"https://avatars0.githubusercontent.com/u/19937171?v=4","gravatar_id":"","url":"https://api.github.com/users/m3db","html_url":"https://github.com/m3db","followers_url":"https://api.github.com/users/m3db/followers","following_url":"https://api.github.com/users/m3db/following{/other_user}","gists_url":"https://api.github.com/users/m3db/gists{/gist_id}","starred_url":"https://api.github.com/users/m3db/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/m3db/subscriptions","organizations_url":"https://api.github.com/users/m3db/orgs","repos_url":"https://api.github.com/users/m3db/repos","events_url":"https://api.github.com/users/m3db/events{/privacy}","received_events_url":"https://api.github.com/users/m3db/received_events","type":"Organization","site_admin":false},"repo":{"id":126348256,"name":"m3msg","full_name":"m3db/m3msg","owner":{"login":"m3db","id":19937171,"avatar_url":"https://avatars0.githubusercontent.com/u/19937171?v=4","gravatar_id":"","url":"https://api.github.com/users/m3db","html_url":"https://github.com/m3db","followers_url":"https://api.github.com/users/m3db/followers","following_url":"https://api.github.com/users/m3db/following{/other_user}","gists_url":"https://api.github.com/users/m3db/gists{/gist_id}","starred_url":"https://api.github.com/users/m3db/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/m3db/subscriptions","organizations_url":"https://api.github.com/users/m3db/orgs","repos_url":"https://api.github.com/users/m3db/repos","events_url":"https://api.github.com/users/m3db/events{/privacy}","received_events_url":"https://api.github.com/users/m3db/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/m3db/m3msg","description":null,"fork":false,"url":"https://api.github.com/repos/m3db/m3msg","forks_url":"https://api.github.com/repos/m3db/m3msg/forks","keys_url":"https://api.github.com/repos/m3db/m3msg/keys{/key_id}","collaborators_url":"https://api.github.com/repos/m3db/m3msg/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/m3db/m3msg/teams","hooks_url":"https://api.github.com/repos/m3db/m3msg/hooks","issue_events_url":"https://api.github.com/repos/m3db/m3msg/issues/events{/number}","events_url":"https://api.github.com/repos/m3db/m3msg/events","assignees_url":"https://api.github.com/repos/m3db/m3msg/assignees{/user}","branches_url":"https://api.github.com/repos/m3db/m3msg/branches{/branch}","tags_url":"https://api.github.com/repos/m3db/m3msg/tags","blobs_url":"https://api.github.com/repos/m3db/m3msg/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/m3db/m3msg/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/m3db/m3msg/git/refs{/sha}","trees_url":"https://api.github.com/repos/m3db/m3msg/git/trees{/sha}","statuses_url":"https://api.github.com/repos/m3db/m3msg/statuses/{sha}","languages_url":"https://api.github.com/repos/m3db/m3msg/languages","stargazers_url":"https://api.github.com/repos/m3db/m3msg/stargazers","contributors_url":"https://api.github.com/repos/m3db/m3msg/contributors","subscribers_url":"https://api.github.com/repos/m3db/m3msg/subscribers","subscription_url":"https://api.github.com/repos/m3db/m3msg/subscription","commits_url":"https://api.github.com/repos/m3db/m3msg/commits{/sha}","git_commits_url":"https://api.github.com/repos/m3db/m3msg/git/commits{/sha}","comments_url":"https://api.github.com/repos/m3db/m3msg/comments{/number}","issue_comment_url":"https://api.github.com/repos/m3db/m3msg/issues/comments{/number}","contents_url":"https://api.github.com/repos/m3db/m3msg/contents/{+path}","compare_url":"https://api.github.com/repos/m3db/m3msg/compare/{base}...{head}","merges_url":"https://api.github.com/repos/m3db/m3msg/merges","archive_url":"https://api.github.com/repos/m3db/m3msg/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/m3db/m3msg/downloads","issues_url":"https://api.github.com/repos/m3db/m3msg/issues{/number}","pulls_url":"https://api.github.com/repos/m3db/m3msg/pulls{/number}","milestones_url":"https://api.github.com/repos/m3db/m3msg/milestones{/number}","notifications_url":"https://api.github.com/repos/m3db/m3msg/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/m3db/m3msg/labels{/name}","releases_url":"https://api.github.com/repos/m3db/m3msg/releases{/id}","deployments_url":"https://api.github.com/repos/m3db/m3msg/deployments","created_at":"2018-03-22T14:32:21Z","updated_at":"2018-04-11T21:06:43Z","pushed_at":"2018-04-12T02:11:39Z","git_url":"git://github.com/m3db/m3msg.git","ssh_url":"git@github.com:m3db/m3msg.git","clone_url":"https://github.com/m3db/m3msg.git","svn_url":"https://github.com/m3db/m3msg","homepage":null,"size":119,"stargazers_count":0,"watchers_count":0,"language":"Go","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":1,"license":{"key":"mit","name":"MIT License","spdx_id":"MIT","url":"https://api.github.com/licenses/mit"},"forks":0,"open_issues":1,"watchers":0,"default_branch":"master"}},"base":{"label":"m3db:master","ref":"master","sha":"6bbff21ed042aef43c6c94ee46f4dcaf1c7b111d","user":{"login":"m3db","id":19937171,"avatar_url":"https://avatars0.githubusercontent.com/u/19937171?v=4","gravatar_id":"","url":"https://api.github.com/users/m3db","html_url":"https://github.com/m3db","followers_url":"https://api.github.com/users/m3db/followers","following_url":"https://api.github.com/users/m3db/following{/other_user}","gists_url":"https://api.github.com/users/m3db/gists{/gist_id}","starred_url":"https://api.github.com/users/m3db/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/m3db/subscriptions","organizations_url":"https://api.github.com/users/m3db/orgs","repos_url":"https://api.github.com/users/m3db/repos","events_url":"https://api.github.com/users/m3db/events{/privacy}","received_events_url":"https://api.github.com/users/m3db/received_events","type":"Organization","site_admin":false},"repo":{"id":126348256,"name":"m3msg","full_name":"m3db/m3msg","owner":{"login":"m3db","id":19937171,"avatar_url":"https://avatars0.githubusercontent.com/u/19937171?v=4","gravatar_id":"","url":"https://api.github.com/users/m3db","html_url":"https://github.com/m3db","followers_url":"https://api.github.com/users/m3db/followers","following_url":"https://api.github.com/users/m3db/following{/other_user}","gists_url":"https://api.github.com/users/m3db/gists{/gist_id}","starred_url":"https://api.github.com/users/m3db/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/m3db/subscriptions","organizations_url":"https://api.github.com/users/m3db/orgs","repos_url":"https://api.github.com/users/m3db/repos","events_url":"https://api.github.com/users/m3db/events{/privacy}","received_events_url":"https://api.github.com/users/m3db/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/m3db/m3msg","description":null,"fork":false,"url":"https://api.github.com/repos/m3db/m3msg","forks_url":"https://api.github.com/repos/m3db/m3msg/forks","keys_url":"https://api.github.com/repos/m3db/m3msg/keys{/key_id}","collaborators_url":"https://api.github.com/repos/m3db/m3msg/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/m3db/m3msg/teams","hooks_url":"https://api.github.com/repos/m3db/m3msg/hooks","issue_events_url":"https://api.github.com/repos/m3db/m3msg/issues/events{/number}","events_url":"https://api.github.com/repos/m3db/m3msg/events","assignees_url":"https://api.github.com/repos/m3db/m3msg/assignees{/user}","branches_url":"https://api.github.com/repos/m3db/m3msg/branches{/branch}","tags_url":"https://api.github.com/repos/m3db/m3msg/tags","blobs_url":"https://api.github.com/repos/m3db/m3msg/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/m3db/m3msg/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/m3db/m3msg/git/refs{/sha}","trees_url":"https://api.github.com/repos/m3db/m3msg/git/trees{/sha}","statuses_url":"https://api.github.com/repos/m3db/m3msg/statuses/{sha}","languages_url":"https://api.github.com/repos/m3db/m3msg/languages","stargazers_url":"https://api.github.com/repos/m3db/m3msg/stargazers","contributors_url":"https://api.github.com/repos/m3db/m3msg/contributors","subscribers_url":"https://api.github.com/repos/m3db/m3msg/subscribers","subscription_url":"https://api.github.com/repos/m3db/m3msg/subscription","commits_url":"https://api.github.com/repos/m3db/m3msg/commits{/sha}","git_commits_url":"https://api.github.com/repos/m3db/m3msg/git/commits{/sha}","comments_url":"https://api.github.com/repos/m3db/m3msg/comments{/number}","issue_comment_url":"https://api.github.com/repos/m3db/m3msg/issues/comments{/number}","contents_url":"https://api.github.com/repos/m3db/m3msg/contents/{+path}","compare_url":"https://api.github.com/repos/m3db/m3msg/compare/{base}...{head}","merges_url":"https://api.github.com/repos/m3db/m3msg/merges","archive_url":"https://api.github.com/repos/m3db/m3msg/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/m3db/m3msg/downloads","issues_url":"https://api.github.com/repos/m3db/m3msg/issues{/number}","pulls_url":"https://api.github.com/repos/m3db/m3msg/pulls{/number}","milestones_url":"https://api.github.com/repos/m3db/m3msg/milestones{/number}","notifications_url":"https://api.github.com/repos/m3db/m3msg/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/m3db/m3msg/labels{/name}","releases_url":"https://api.github.com/repos/m3db/m3msg/releases{/id}","deployments_url":"https://api.github.com/repos/m3db/m3msg/deployments","created_at":"2018-03-22T14:32:21Z","updated_at":"2018-04-11T21:06:43Z","pushed_at":"2018-04-12T02:11:39Z","git_url":"git://github.com/m3db/m3msg.git","ssh_url":"git@github.com:m3db/m3msg.git","clone_url":"https://github.com/m3db/m3msg.git","svn_url":"https://github.com/m3db/m3msg","homepage":null,"size":119,"stargazers_count":0,"watchers_count":0,"language":"Go","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":1,"license":{"key":"mit","name":"MIT License","spdx_id":"MIT","url":"https://api.github.com/licenses/mit"},"forks":0,"open_issues":1,"watchers":0,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/m3db/m3msg/pulls/8"},"html":{"href":"https://github.com/m3db/m3msg/pull/8"},"issue":{"href":"https://api.github.com/repos/m3db/m3msg/issues/8"},"comments":{"href":"https://api.github.com/repos/m3db/m3msg/issues/8/comments"},"review_comments":{"href":"https://api.github.com/repos/m3db/m3msg/pulls/8/comments"},"review_comment":{"href":"https://api.github.com/repos/m3db/m3msg/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/m3db/m3msg/pulls/8/commits"},"statuses":{"href":"https://api.github.com/repos/m3db/m3msg/statuses/f23549c5fe6da89f0edbd2235d4916c39e35c1b6"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"xichen2020"}}