Ostatnio czytałem o projektowaniu języka Go i ten post jest zbiorem uwag specyficznych dla buforowanych kanałów i ich działania.

Jeśli chcesz zrozumieć, czym jest kanał w go, zapoznaj się z tym artykułem [1] . Po tym punkcie zakładam, że mamy podstawową wiedzę na temat kanału go, to powiedziawszy, przejdźmy teraz do rzeczy!

Tworzenie kanału Go

Zanim zagłębimy się w to, jak powstają kanały go, zobaczmy, jak wyglądają od spodu.

Rzuć okiem na strukturę „hchan”.

Powyższy fragment pochodzi z go/src/runtime/chan.go, pliku zawierającego implementację kanałów go. Struktura hchanto taka, która przechowuje wszystkie informacje o danym kanale.

hchanma element buf, który przechowuje dane jako kolejkę cykliczną. sendxjest indeksem, w którym goroutines wypchnie następny element, podczas gdy recvxjest indeksem, z którego zostanie pobrany następny element.

Ważną rzeczą do odnotowania jest to, że locknie tylko chroni pola w hchan, ale także chroni pola gorutyn ( sudog) zablokowanych w kanałach.

Teraz, gdy rozumiemy podstawową strukturę kanału, przyjrzyjmy się, jak są one inicjowane.

Funkcja makechan

// go/src/runtime/chan.go

func makechan(t *chantype, size int) *hchan {
   elem := t.elem

   // compiler checks this but be safe.
   if elem.size >= 1<<16 {
      throw("makechan: invalid channel element type")
   }
   if hchanSize%maxAlign != 0 || elem.align > maxAlign {
      throw("makechan: bad alignment")
   }

   mem, overflow := math.MulUintptr(elem.size, uintptr(size))
   if overflow || mem > maxAlloc-hchanSize || size < 0 {
      panic(plainError("makechan: size out of range"))
   }

   // Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.
   // buf points into the same allocation, elemtype is persistent.
   // SudoG's are referenced from their owning thread so they can't be collected.
   // TODO(dvyukov,rlh): Rethink when collector can move allocated objects.
   var c *hchan
   switch {
   case mem == 0:
      // Queue or element size is zero.
      c = (*hchan)(mallocgc(hchanSize, nil, true))
      // Race detector uses this location for synchronization.
      c.buf = c.raceaddr()
   case elem.ptrdata == 0:
      // Elements do not contain pointers.
      // Allocate hchan and buf in one call.
      c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
      c.buf = add(unsafe.Pointer(c), hchanSize)
   default:
      // Elements contain pointers.
      c = new(hchan)
      c.buf = mallocgc(mem, elem, true)
   }

   c.elemsize = uint16(elem.size)
   c.elemtype = elem
   c.dataqsiz = uint(size)
   lockInit(&c.lock, lockRankHchan)

   if debugChan {
      print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
   }
   return c
}

Będę tutaj szczery, nie jestem pewien, o co chodzi w walidacji złego wyrównania, nurkowanie trochę głębiej pokazuje, że chodzi o walidację pamięci i wyrównanie typów w go, kilka tematów, co do których nie jestem pewien. Pewnie następny wpis na blogu będzie o tym?

Wysyłanie elementu do kanału

Porozmawiajmy teraz o tym, jak goroutine ( g1 ) popycha element w kanale.

Wysyłki kanałów są obsługiwane przezchansend

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  if c == nil {
    if !block {
     return false
    }
    gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
    throw("unreachable")
   }
  
   if debugChan {
    print("chansend: chan=", c, "\n")
   }
  
   if raceenabled {
    racereadpc(c.raceaddr(), callerpc, abi.FuncPCABIInternal(chansend))
   }
  
   // Fast path: check for failed non-blocking operation without acquiring the lock.
   //
   // After observing that the channel is not closed, we observe that the channel is
   // not ready for sending. Each of these observations is a single word-sized read
   // (first c.closed and second full()).
   // Because a closed channel cannot transition from 'ready for sending' to
   // 'not ready for sending', even if the channel is closed between the two observations,
   // they imply a moment between the two when the channel was both not yet closed
   // and not ready for sending. We behave as if we observed the channel at that moment,
   // and report that the send cannot proceed.
   //
   // It is okay if the reads are reordered here: if we observe that the channel is not
   // ready for sending and then observe that it is not closed, that implies that the
   // channel wasn't closed during the first observation. However, nothing here
   // guarantees forward progress. We rely on the side effects of lock release in
   // chanrecv() and closechan() to update this thread's view of c.closed and full().
   if !block && c.closed == 0 && full(c) {
    return false
   }
  ...
}

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  ...
  if sg := c.recvq.dequeue(); sg != nil {
     // Found a waiting receiver. We pass the value we want to send
     // directly to the receiver, bypassing the channel buffer (if any).
     send(c, sg, ep, func() { unlock(&c.lock) }, 3)
     return true
  }
  ...
}

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  ...
  if c.qcount < c.dataqsiz {
     // Space is available in the channel buffer. Enqueue the element to send.
     qp := chanbuf(c, c.sendx)
     if raceenabled {
        racenotify(c, c.sendx, nil)
     }
     typedmemmove(c.elemtype, qp, ep)
     c.sendx++
     if c.sendx == c.dataqsiz {
        c.sendx = 0
     }
     c.qcount++
     unlock(&c.lock)
     return true
  }
  ...
}

Ale… co się stanie, jeśli zabraknie miejsca w buforze kanału? Tutaj mogą wystąpić dwie sytuacje.

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  
  ...
  
  if !block {
     unlock(&c.lock)
     return false
  }
  
  // Block on the channel. Some receiver will complete our operation for us.
  gp := getg()
  mysg := acquireSudog()
  mysg.releasetime = 0
  if t0 != 0 {
     mysg.releasetime = -1
  }
  // No stack splits between assigning elem and enqueuing mysg
  // on gp.waiting where copystack can find it.
  mysg.elem = ep
  mysg.waitlink = nil
  mysg.g = gp
  mysg.isSelect = false
  mysg.c = c
  gp.waiting = mysg
  gp.param = nil
  c.sendq.enqueue(mysg)
  // Signal to anyone trying to shrink our stack that we're about
  // to park on a channel. The window between when this G's status
  // changes and when we set gp.activeStackChans is not safe for
  // stack shrinking.
  gp.parkingOnChan.Store(true)
  gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
  // Ensure the value being sent is kept alive until the
  // receiver copies it out. The sudog has a pointer to the
  // stack object, but sudogs aren't considered as roots of the
  // stack tracer.
  KeepAlive(ep)
  
  // someone woke us up.
  if mysg != gp.waiting {
     throw("G waiting list is corrupted")
  }
  gp.waiting = nil
  gp.activeStackChans = false
  closed := !mysg.success
  gp.param = nil
  if mysg.releasetime > 0 {
     blockevent(mysg.releasetime-t0, 2)
  }
  mysg.c = nil
  releaseSudog(mysg)
  if closed {
     if c.closed == 0 {
        throw("chansend: spurious wakeup")
     }
     panic(plainError("send on closed channel"))
  }
  return true
}

Sudog tutaj jest strukturą, która przechowuje informacje dla oczekującej gorutyny (w naszym przypadku g1 ) . Go runtime zaparkuje tę wysyłającą goroutine (g1) i zgodnie z komentarzami zapewniamy to

1. Stos g1 nie powinien się zmniejszać — (więcej o zwiększaniu i zmniejszaniu stosu można dowiedzieć się z tego filmu [4])
2. Upewnij się, że wartość, którą g1 miał wysłać, jest utrzymywana przy życiu, dopóki odbiorca (g2) nie skopiuje jej.

Kiedy nadchodzi goroutine odbiornika ( g2 ) i nas budzi, w końcu puszczamy sudogi jeszcze raz sprawdzamy, czy kanał był zamknięty, czy nie. Jeśli tak, zgłoś błąd, w przeciwnym razie zwróć prawdę!

A teraz, o co chodziło z tym bezpośrednim wysyłaniem?

W pustym/niebuforowanym kanale, gdy g2 jest już obecne w recvq, go pozwala g1 bezpośrednio skopiować element w odbierającym stosie g2 .

Go GC zakłada, że ​​zapisy stosu mają miejsce tylko wtedy, gdy działa g1 i są wykonywane tylko przez niego.

Należy zauważyć, że bariera zapisu jest ustanawiana przed skopiowaniem elementu na stos odbierającej gorutyny. Oszczędza to jedną operację kopiowania pamięci! Rzeczywistą implementację memmovemożna znaleźć tutaj — CPP Reference [5]

Odbiór elementu z kanału

Jak sugeruje tytuł, tutaj musimy zrozumieć, w jaki sposób goroutine otrzymuje element z kanału. zgodnie z wcześniejszymi konwencjami, odbiornikiem goroutine będzie g2, a wysyłającym goroutine będzie g1.

Odbiory są obsługiwane przez chanrecv, a dokumentacja pięknie wyjaśniła jego rolę, więc powiem tylko, że tutaj —

chanrecv odbiera na kanale c i zapisuje odebrane dane do ep .
ep może być zerowe, w takim przypadku odebrane dane są ignorowane.
Jeśli block == false i żadne elementy nie są dostępne, zwraca (false, false).
W przeciwnym razie, jeśli c jest domknięte, zera * ep i zwraca (prawda, fałsz).
W przeciwnym razie wypełnia * ep elementem i zwraca (true, true). Odp
różny od zera musi wskazywać na stertę lub stos wywołującego.

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
  ...
  if c.qcount > 0 {
     // Receive directly from queue
     qp := chanbuf(c, c.recvx)
     if raceenabled {
        racenotify(c, c.recvx, nil)
     }
     if ep != nil {
        typedmemmove(c.elemtype, ep, qp)
     }
     typedmemclr(c.elemtype, qp)
     c.recvx++
     if c.recvx == c.dataqsiz {
        c.recvx = 0
     }
     c.qcount--
     unlock(&c.lock)
     return true, true
  }
  ...
}

Jeśli g1 już było obecne w sendq, g2 zainicjuje bezpośrednie odbieranie do, epjeśli bufor jest pusty, tak jak bezpośrednie wysyłanie zostało wykonane, w przeciwnym razie g2 pobierze element z nagłówka bufora i skopiuje element z g1 ' s stos do końca bufora kanału i odparkuj g1. (tj. ustaw g1 z powrotem na runnable)

// The portion responsible to initiate the 2 phase receive
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
   if c.dataqsiz == 0 {
      if raceenabled {
         racesync(c, sg)
      }
      if ep != nil {
         // copy data from sender
         recvDirect(c.elemtype, sg, ep)
      }
   } else {
      // Queue is full. Take the item at the
      // head of the queue. Make the sender enqueue
      // its item at the tail of the queue. Since the
      // queue is full, those are both the same slot.
      qp := chanbuf(c, c.recvx)
      if raceenabled {
         racenotify(c, c.recvx, nil)
         racenotify(c, c.recvx, sg)
      }
      // copy data from queue to receiver
      if ep != nil {
         typedmemmove(c.elemtype, ep, qp)
      }
      // copy data from sender to queue
      typedmemmove(c.elemtype, qp, sg.elem)
      c.recvx++
      if c.recvx == c.dataqsiz {
         c.recvx = 0
      }
      c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
   }
   sg.elem = nil
   gp := sg.g
   unlockf()
   gp.param = unsafe.Pointer(sg)
   sg.success = true
   if sg.releasetime != 0 {
      sg.releasetime = cputicks()
   }
   goready(gp, skip+1)
}

Wreszcie, jeśli nie ma żadnego elementu do odebrania z kanału, kanał albo się odblokuje i zwróci wartość true; lub zaparkuj g2 i ustaw go w kolejce recvq, aż pojawi się nadawca ( g1 ) i obudzi g2 .

Wniosek

Mam nadzieję, że dzięki tym notatkom i towarzyszącym im ilustracjom udało mi się wyjaśnić, jak wewnętrzne kanały działają w Go.

Pokochałem małe poprawki i optymalizacje, które inżynierowie Go wprowadzili w tak prostej strukturze danych, aby zaoszczędzić kilka cykli procesora i zapewnić bezpieczeństwo pamięci, takie jak bezpośrednie wysyłanie i odbieranie , memcpy i mam nadzieję, że zilustrowałem je wystarczająco dobrze dla ciebie żeby je docenić :D

Zasoby

• [1] Przejdź do kanałów —https://gobyexample.com/channels
• [2] GopherCon 2017: Kavya Joshi — Zrozumienie kanałów
• [3] Układ pamięci w Go
• [4] Harmonogram Go: język implementacji z lekką współbieżnością
• [5] Odniesienie do CPP — memmove_.s

Dziękujemy za bycie częścią naszej społeczności! Zanim pójdziesz:

• Klaskajcie za relację i śledźcie autora
• Zobacz więcej treści w publikacji Level Up Coding
• BEZPŁATNY kurs rozmowy kwalifikacyjnej z kodowania
• Śledź nas: Twitter | LinkedIn | Biuletyn