Golang how can I guarantee data finishes in a goroutine while being accessed by another goroutine

huangapple
go评论106阅读模式
英文:

Golang how can I guarantee data finishes in a goroutine while being accessed by another goroutine

问题

大家好，我正在尝试使用Websockets创建一个派对系统，人们可以进入队列，然后与5个与他们相似的人匹配。目前我在以下部分遇到了问题：

  1. type PartyHub struct {
  2.     Partys             map[string]*Party
  3.     PartialPartys      []*PartialParty
  4.     Queue              []*Member
  5.     AddParty           chan *Party
  6.     RemoveParty        chan *Party
  7.     AddPartialParty    chan *PartialParty
  8.     RemovePartialParty chan *PartialParty
  9.     EnterQueue         chan *Member
  10.     LeaveQueue         chan *Member
  11.     Mu                 sync.Mutex
  12. }
  14. func (p *PartyHub) Run() {
  15.     for {
  16.         select {
  17.         case member := <-p.EnterQueue:
  18.             go p.SortMemberIntoParty(member)
  19.             go p.SortMemberIntoParty(member)
  20.             go p.SortMemberIntoParty(member)
  21.             go p.SortMemberIntoParty(member)
  22.             go p.SortMemberIntoParty(member)
  23.             
  24.             log.Println(p.PartialPartys)
  25.         case party := <-p.AddPartialParty:
  26.             p.Mu.Lock()
  27.             defer p.Mu.Unlock()
  28.             p.PartialPartys = append(p.PartialPartys, party)
  29.         }
  30.     }
  31. }
  33. func (p *PartyHub) SortMemberIntoParty(member *Member) {
  34.     p.Mu.Lock()
  35.     defer p.Mu.Unlock()
  36.     if len(p.PartialPartys) == 0 {
  37.         log.Println("Here")
  38.         newParty := &PartialParty{Accepting: true, Members: []*Member{member}}
  39.         p.AddPartialParty <- newParty
  40.         return
  41.     }
  43.     foundPartyForMember := false
  44.     for _, party := range p.PartialPartys {
  45.         goodFitForParty := true
  46.         for _, partyMember := range party.Members {
  47.             log.Println(member.Type == partyMember.Type, member.Rank >= partyMember.Rank-partyMember.RankTol, member.Rank <= partyMember.Rank+partyMember.RankTol)
  48.             if member.Type == partyMember.Type && member.Rank >= partyMember.Rank-partyMember.RankTol && member.Rank <= partyMember.Rank+partyMember.RankTol {
  49.                 
  50.                 goodFitForParty = true
  51.                 continue
  52.             } else {
  53.                 goodFitForParty = false
  54.                 break
  55.             }
  56.         }
  58.         if !goodFitForParty {
  59.             continue
  60.         } else {
  61.             foundPartyForMember = true
  62.             party.Mu.Lock()
  63.             defer party.Mu.Unlock()
  64.             party.Members = append(party.Members, member)
  65.             if len(party.Members) == 5 {
  66.                 party.Accepting = false
  67.                 go party.SendReadyCheck()
  68.             }
  69.             break
  70.         }
  71.     }
  72.     
  74.     if !foundPartyForMember {
  75.         newParty := &PartialParty{Accepting: true, Members: []*Member{member}}
  76.         p.AddPartialParty <- newParty
  77.     }
  78.     
  79.     log.Println("Sorting Members")
  80. }

唯一的问题是，这5个goroutine似乎比数据更快地完成了操作。

例如：p.PartialPartys显示没有派对。

我需要的是，对于每个访问PartyHub结构体的goroutine，p.PartialPartys始终保持最新状态。我以为sync.Mutex会为我完成这个工作，但似乎并不是这样。有人能告诉我保持所有goroutine与相同数据同步的最佳方法吗？

英文:

Hi guys I am trying to make a party system using websockets where people can enter a queue and then get matched with 5 people similar to them. Right now I'm having trouble with this part:

  1. type PartyHub struct {
  2. Partys             map[string]*Party
  3. PartialPartys      []*PartialParty
  4. Queue              []*Member
  5. AddParty           chan *Party
  6. RemoveParty        chan *Party
  7. AddPartialParty    chan *PartialParty
  8. RemovePartialParty chan *PartialParty
  9. EnterQueue         chan *Member
  10. LeaveQueue         chan *Member
  11. Mu                 sync.Mutex
  12. }
  13. // Run will begin monitoring the channels
  14. // to register and unregister partys as they are
  15. // created or destroyed
  16. func (p *PartyHub) Run() {
  17. for {
  18. select {
  19. case member := &lt;-p.EnterQueue:
  20. go p.SortMemberIntoParty(member)
  21. go p.SortMemberIntoParty(member)
  22. go p.SortMemberIntoParty(member)
  23. go p.SortMemberIntoParty(member)
  24. go p.SortMemberIntoParty(member)
  25. log.Println(p.PartialPartys)
  26. case party := &lt;-p.AddPartialParty:
  27. p.Mu.Lock()
  28. defer p.Mu.Unlock()
  29. p.PartialPartys = append(p.PartialPartys, party)
  30. }
  31. }
  32. }
  33. // SortMemberIntoParty will take a new user entering the queue and find an appropriate Party
  34. // for the member to join, taking into account RankTollerance, Rank
  35. func (p *PartyHub) SortMemberIntoParty(member *Member) {
  36. p.Mu.Lock()
  37. defer p.Mu.Unlock()
  38. if len(p.PartialPartys) == 0 {
  39. log.Println(&quot;Here&quot;)
  40. newParty := &amp;PartialParty{Accepting: true, Members: []*Member{member}}
  41. p.AddPartialParty &lt;- newParty
  42. return
  43. }
  44. foundPartyForMember := false
  45. for _, party := range p.PartialPartys {
  46. goodFitForParty := true
  47. for _, partyMember := range party.Members {
  48. log.Println(member.Type == partyMember.Type, member.Rank &gt;= partyMember.Rank-partyMember.RankTol, member.Rank &lt;= partyMember.Rank+partyMember.RankTol)
  49. if member.Type == partyMember.Type &amp;&amp; member.Rank &gt;= partyMember.Rank-partyMember.RankTol &amp;&amp; member.Rank &lt;= partyMember.Rank+partyMember.RankTol {
  50. goodFitForParty = true
  51. continue
  52. } else {
  53. goodFitForParty = false
  54. break
  55. }
  56. }
  57. if !goodFitForParty {
  58. continue
  59. } else {
  60. foundPartyForMember = true
  61. party.Mu.Lock()
  62. defer party.Mu.Unlock()
  63. party.Members = append(party.Members, member)
  64. if len(party.Members) == 5 {
  65. party.Accepting = false
  66. go party.SendReadyCheck()
  67. }
  68. break
  69. }
  70. }
  71. if !foundPartyForMember {
  72. newParty := &amp;PartialParty{Accepting: true, Members: []*Member{member}}
  73. p.AddPartialParty &lt;- newParty
  74. }
  75. log.Println(&quot;Sorting Members&quot;)
  76. }

The only problem is, the 5 goroutines seem to finish quicker than the data knows what happened.

For example: p.PartialPartys says it has no parties.

What I need is to have p.PartialPartys always up to date for every goroutine that accesses that field of the PartyHub struct I though the sync.Mutex would do this for me but it doesn't seem to be the case, can anybody tell me the best way to keep all my goroutines in sync with the same data?

答案1

得分: 5

所以根据这个实现，你的五个goroutine都无法并行运行，因为它们都试图获取p.Mu互斥锁。从你使用p.AddPartialParty通道的方式来看，如果代码发生死锁，我也不会感到惊讶。

考虑以下事件序列：

  1. 其中一个SortMemberIntoParty goroutine开始运行并获取互斥锁。
  2. 它在p.AddPartialParty上发送一个值，被Run接收。然后，Run尝试获取互斥锁，因此阻塞。
  3. 原始的SortMemberIntoParty goroutine完成并释放互斥锁。
  4. 另一个SortMemberIntoParty goroutine获取互斥锁，并尝试向p.AddPartialParty发送另一个值。
  5. 该goroutine被阻塞，因为没有人准备读取该值（Run仍在等待互斥锁，然后才能回到select语句）。

现在你有一个被阻塞的goroutine持有一个被通道接收端需要的锁。还要注意，在（4）中，你看不到新的PartialParty，因为Run还没有来得及添加它。

如果确实需要互斥锁，那么直接让你的SortMemberIntoParty goroutine直接更新p.PartialPartys可能更容易，而不是使用通道：你已经知道没有其他人会同时访问该变量。

还值得记住的是，这个互斥锁实际上意味着所有的SortMemberIntoParty goroutine将被串行化。如果你希望在这里使用goroutine实现并行性，那么互斥锁将破坏这一点。

英文:

So with this implementation, none of your five goroutines are going to be be able to run in parallel because they are all trying to acquite the p.Mu mutex. And looking at the way you're using the p.AddPartialParty channel, I wouldn't be surprised if the code can deadlock.

Consider the following sequence of events:

  1. One of the SortMemberIntoParty goroutines starts running and acquires the mutex.
  2. It sends a value on p.AddPartialParty, which is received by Run. Run then tries to acquire the mutex, so blocks.
  3. The original SortMemberIntoParty goroutine completes and releases the mutex.
  4. A different SortMemberIntoParty goroutine acquires the mutex, and tries to send another value to p.AddPartialParty.
  5. The goroutine blocks because there is no one ready to read the value (Run is still waiting on the mutex before it gets back to the select statement).

So now you've got a blocked goroutine holding a lock needed by the receiving end of the channel. Also note that at (4) you won't see the new PartialParty because Run hasn't managed to add it yet.

If you do need the mutex, then it might be easier to just have your SortMemberIntoParty goroutine update p.PartialPartys directly rather than using the channel: you already know that no one else will be accessing the variable concurrently.

It's also worth remembering that this mutex essentially means that all SortMemberIntoParty goroutines will be serialised. If you were using goroutines in the hope of achieving parallelism here, the mutex defeats that.

huangapple
  • 本文由 发表于 2015年12月1日 09:33:34
  • 转载请务必保留本文链接：https://go.coder-hub.com/34011131.html
  • go
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定