// JoinGroup collects a number of nodes to join and sends join requests to them.
// nodesJoined contains the nodes that are already joined. The method returns
// the nodes that are succesfully joined, and the nodes that failed respond.
438471
func (j *joinSender) JoinGroup(nodesJoined []string) ([]string, []string) {
	group := j.SelectGroup(nodesJoined)

	var responses struct {
		successes []string
		failures  []string
		sync.Mutex
	}

	var numNodesLeft = j.size - len(nodesJoined)
	var startTime = time.Now()

	var wg sync.WaitGroup

	j.numTries++
	j.node.EmitEvent(JoinTriesUpdateEvent{j.numTries})

	for _, target := range group {
		wg.Add(1)

		go func(target string) {
			defer wg.Done()

			res, err := sendJoinRequest(j.node, target, j.timeout)

			if err != nil {
				msg := "attempt to join node failed"
				if err == errJoinTimeout {
					msg = "attempt to join node timed out"
				}

				j.logger.WithFields(log.Fields{
					"remote":  target,
					"timeout": j.timeout,
				}).Debug(msg)

				responses.Lock()
				responses.failures = append(responses.failures, target)
				responses.Unlock()
				return
			}

			responses.Lock()
			responses.successes = append(responses.successes, target)
			responses.Unlock()

			start := time.Now()
			j.node.memberlist.AddJoinList(res.Membership)

			j.node.EmitEvent(AddJoinListEvent{
				Duration: time.Now().Sub(start),
			})
		}(target)
	}

	// wait for joins to complete
	wg.Wait()

	// don't need to lock successes/failures since we're finished writing to them
	j.logger.WithFields(log.Fields{
		"groupSize":    len(group),
		"joinSize":     j.size,
		"joinTime":     time.Now().Sub(startTime),
		"numNodesLeft": numNodesLeft,
		"numFailures":  len(responses.failures),
		"failures":     responses.failures,
		"numSuccesses": len(responses.successes),
		"successes":    responses.successes,
	}).Debug("join group complete")

	return responses.successes, responses.failures
}
32153
func (j *joinSender) SelectGroup(nodesJoined []string) []string {
	var group []string
	// if fully exhausted or first round, initialize this round's nodes
	if len(j.roundPreferredNodes) == 0 && len(j.roundNonPreferredNodes) == 0 {
		j.Init(nodesJoined)
	}

	numNodesLeft := j.size - len(nodesJoined)

	cont := func() bool {
		if len(group) == numNodesLeft*j.parallelismFactor {
			return false
		}

		nodesAvailable := len(j.roundPreferredNodes) + len(j.roundNonPreferredNodes)
		if nodesAvailable == 0 {
			return false
		}

		return true
	}

	for cont() {
		if len(j.roundPreferredNodes) > 0 {
			group = append(group, util.TakeNode(&j.roundPreferredNodes, -1))
		} else if len(j.roundNonPreferredNodes) > 0 {
			group = append(group, util.TakeNode(&j.roundNonPreferredNodes, -1))
		}
	}

	return group
}
11709
func (n *RequestObserver) Join(req *http.Request, connectionID uint64) {
	n.state.id = connectionID
	n.state.websocketConnected = n.clock.Now()

	n.logger.Debugf(
		"[%X] API connection from %s",
		n.state.id,
		req.RemoteAddr,
	)
}
925285
func Joiner(first, others func(string) string) func(pre string, in []string, post string) string {
	return func(pre string, in []string, post string) string {
		tmp := []string{others(pre)}
		for i := range in {
			tmp = append(tmp, others(in[i]))
		}
		tmp = append(tmp, others(post))
		return first(strings.Join(tmp, ""))
	}
}
1326751
func (group *NodeGroup) Nodes() []Node {
	if group == nil {
		return nil
	}

	group.mux.RLock()
	defer group.mux.RUnlock()

	return group.nodes
}
493671
func (s *ChannelsService) Join(name string) *ChannelsJoinCall {
	var call ChannelsJoinCall
	call.service = s
	call.name = name
	return &call
}
1127358
func (this *Dataset) CoGroupPartitionedSorted(name string, that *Dataset, indexes []int) (ret *Dataset) {
	ret = this.Flow.NewNextDataset(len(this.Shards))
	ret.IsPartitionedBy = indexes

	inputs := []*Dataset{this, that}
	step := this.Flow.MergeDatasets1ShardTo1Step(inputs, ret)
	step.SetInstruction(name, instruction.NewCoGroupPartitionedSorted(indexes))
	return ret
}
25907
func (b *Builder) InnerJoin(joinTable string, joinCond interface{}) *Builder {
	return b.Join("INNER", joinTable, joinCond)
}
384712
func (d *Dataset) Join(name string, other *Dataset, sortOption *SortOption) *Dataset {
	return d.DoJoin(name, other, false, false, sortOption)
}
629716
func (n *BuilderNode) Group() *BuilderGroup {
	return n.builder.groups[n.builder.ring.NodeToGroup[n.index]]
}
871960
func (s *Stack) JoinGroup(protocol tcpip.NetworkProtocolNumber, nicID tcpip.NICID, multicastAddr tcpip.Address) *tcpip.Error {
	// TODO: notify network of subscription via igmp protocol.
	return s.AddAddressWithOptions(nicID, protocol, multicastAddr, NeverPrimaryEndpoint)
}
172717
func (d *driver) Join(nid, eid string, sboxKey string, jinfo driverapi.JoinInfo, options map[string]interface{}) error {
	defer osl.InitOSContext()()

	network, err := d.getNetwork(nid)
	if err != nil {
		return err
	}

	endpoint, err := network.getEndpoint(eid)
	if err != nil {
		return err
	}

	if endpoint == nil {
		return EndpointNotFoundError(eid)
	}

	endpoint.containerConfig, err = parseContainerOptions(options)
	if err != nil {
		return err
	}

	iNames := jinfo.InterfaceName()
	containerVethPrefix := defaultContainerVethPrefix
	if network.config.ContainerIfacePrefix != "" {
		containerVethPrefix = network.config.ContainerIfacePrefix
	}
	err = iNames.SetNames(endpoint.srcName, containerVethPrefix)
	if err != nil {
		return err
	}

	err = jinfo.SetGateway(network.bridge.gatewayIPv4)
	if err != nil {
		return err
	}

	err = jinfo.SetGatewayIPv6(network.bridge.gatewayIPv6)
	if err != nil {
		return err
	}

	return nil
}
834327
func (d *driver) Join(nid, eid string, sboxKey string, jinfo driverapi.JoinInfo, options map[string]interface{}) error {
	defer osl.InitOSContext()()
	n, err := d.getNetwork(nid)
	if err != nil {
		return err
	}
	endpoint := n.endpoint(eid)
	if endpoint == nil {
		return fmt.Errorf("could not find endpoint with id %s", eid)
	}
	// generate a name for the iface that will be renamed to eth0 in the sbox
	containerIfName, err := netutils.GenerateIfaceName(ns.NlHandle(), vethPrefix, vethLen)
	if err != nil {
		return fmt.Errorf("error generating an interface name: %s", err)
	}
	// create the netlink macvlan interface
	vethName, err := createMacVlan(containerIfName, n.config.Parent, n.config.MacvlanMode)
	if err != nil {
		return err
	}
	// bind the generated iface name to the endpoint
	endpoint.srcName = vethName
	ep := n.endpoint(eid)
	if ep == nil {
		return fmt.Errorf("could not find endpoint with id %s", eid)
	}
	// parse and match the endpoint address with the available v4 subnets
	if len(n.config.Ipv4Subnets) > 0 {
		s := n.getSubnetforIPv4(ep.addr)
		if s == nil {
			return fmt.Errorf("could not find a valid ipv4 subnet for endpoint %s", eid)
		}
		v4gw, _, err := net.ParseCIDR(s.GwIP)
		if err != nil {
			return fmt.Errorf("gateway %s is not a valid ipv4 address: %v", s.GwIP, err)
		}
		err = jinfo.SetGateway(v4gw)
		if err != nil {
			return err
		}
		logrus.Debugf("Macvlan Endpoint Joined with IPv4_Addr: %s, Gateway: %s, MacVlan_Mode: %s, Parent: %s",
			ep.addr.IP.String(), v4gw.String(), n.config.MacvlanMode, n.config.Parent)
	}
	// parse and match the endpoint address with the available v6 subnets
	if len(n.config.Ipv6Subnets) > 0 {
		s := n.getSubnetforIPv6(ep.addrv6)
		if s == nil {
			return fmt.Errorf("could not find a valid ipv6 subnet for endpoint %s", eid)
		}
		v6gw, _, err := net.ParseCIDR(s.GwIP)
		if err != nil {
			return fmt.Errorf("gateway %s is not a valid ipv6 address: %v", s.GwIP, err)
		}
		err = jinfo.SetGatewayIPv6(v6gw)
		if err != nil {
			return err
		}
		logrus.Debugf("Macvlan Endpoint Joined with IPv6_Addr: %s Gateway: %s MacVlan_Mode: %s, Parent: %s",
			ep.addrv6.IP.String(), v6gw.String(), n.config.MacvlanMode, n.config.Parent)
	}
	iNames := jinfo.InterfaceName()
	err = iNames.SetNames(vethName, containerVethPrefix)
	if err != nil {
		return err
	}
	if err := d.storeUpdate(ep); err != nil {
		return fmt.Errorf("failed to save macvlan endpoint %.7s to store: %v", ep.id, err)
	}
	return nil
}
478295
func (w *Watcher) Join(client Client, recursive bool) *probe.Error {
	wo, err := client.Watch(watchParams{
		recursive: recursive,
		events:    []string{"put", "delete"},
	})
	if err != nil {
		return err
	}

	w.o = append(w.o, wo)

	// join monitoring waitgroup
	w.wg.Add(1)

	// wait for events and errors of individual client watchers
	// and sent then to eventsChan and errorsChan
	go func() {
		defer w.wg.Done()

		for {
			select {
			case event, ok := <-wo.Events():
				if !ok {
					return
				}

				w.eventInfoChan <- event
			case err, ok := <-wo.Errors():
				if !ok {
					return
				}

				w.errorChan <- err
			}
		}
	}()

	return nil
}
212007
func (rfs *RootFS) Join(elem ...string) string {
	return path.Clean(path.Join(elem...))
}
166363
func (n *BuilderNode) SetGroup(group *BuilderGroup) {
	n.builder.ring.NodeToGroup[n.index] = group.index
}
1307892
func (b *Broker) JoinGroup(request *JoinGroupRequest) (*JoinGroupResponse, error) {
	response := new(JoinGroupResponse)

	err := b.sendAndReceive(request, response)
	if err != nil {
		return nil, err
	}

	return response, nil
}