func (t *tagManager) Channel(tag int) chan []byte {
	t.Mux.Lock()
	defer t.Mux.Unlock()
	_, ok := t.CommMap[tag]
	if !ok {
		panic("attempt to return chan from non-existant tag")
	}
	c := t.CommMap[tag]
	return c
}
Channel returns the channel for that tag
package mpi
import (
	"bytes"
	"encoding/gob"
	"errors"
	"fmt"
	"net"
	"sort"
	"sync"
	"time"
)
type Network struct {
	NetProto string         
	Addr     string         
	Addrs    []string       
	Timeout  time.Duration  
	Password       string
	hashedPassword string
	myrank int  
	nNodes int  
	connections []*pairwiseConnection  
	local       *local
}
func (n *Network) Rank() int {
	if n.nNodes == 0 {
		return -1
	}
	return n.myrank
}
func (n *Network) Size() int {
	return n.nNodes
}
func (n *Network) Init() error {
	n.useFlags()
	if len(n.Addrs) == 0 {
		n.Addr = ":5000"
		n.Addrs = []string{":5000"}
	}
	err := n.assignRanks()
	if err != nil {
		return err
	}
	return n.startConnections()
}
func (n *Network) useFlags() {
	if n.NetProto == "" {
		n.NetProto = FlagProtocol
	}
	if n.Password == "" {
		n.Password = FlagPassword
	}
	if n.Timeout == 0 {
		n.Timeout = time.Duration(FlagInitTimeout)
	}
	if n.Addr == "" {
		n.Addr = FlagAddr
	}
	if len(n.Addrs) == 0 {
		n.Addrs = make([]string, len(FlagAllAddrs))
		for i, str := range FlagAllAddrs {
			n.Addrs[i] = str
		}
	}
	n.hashedPassword = n.Password  
}
func (n *Network) assignRanks() error {
	sort.Strings(n.Addrs)
	if !uniqueAddrs(n.Addrs) {
		return fmt.Errorf("network addresses not unique. list is: ", n.Addrs)
	}
	n.myrank = sort.SearchStrings(n.Addrs, n.Addr)
	if !(n.myrank < len(n.Addrs) && n.Addrs[n.myrank] == n.Addr) {
		return fmt.Errorf("mpi init: local ip address not in global list. Local address is: %v, global list is %v", n.myrank, n.Addrs)
	}
	n.nNodes = len(n.Addrs)
	return nil
}
func uniqueAddrs(addrs []string) bool {
	for i := 0; i < len(addrs)-1; i++ {
		if addrs[i] == addrs[i+1] {
			return false
		}
	}
	return true
}
func (n *Network) startConnections() error {
	n.connections = make([]*pairwiseConnection, n.nNodes)
	for i := range n.connections {
		n.connections[i] = &pairwiseConnection{
			receivetags: newTagManager(),
			sendtags:    newTagManager(),
		}
	}
	n.local = newLocal()
	var listenError error
	var dialError error
	wg := &sync.WaitGroup{}
	wg.Add(2)
	go func() {
		defer wg.Done()
		listenError = n.establishListenConnections()
	}()
	go func() {
		defer wg.Done()
		dialError = n.establishDialConnections()
	}()
	wg.Wait()
	if listenError != nil {
		return listenError
	}
	if dialError != nil {
		return dialError
	}
	return nil
}
func (n *Network) establishListenConnections() error {
	listener, err := net.Listen(n.NetProto, n.Addr)
	if err != nil {
		return errors.New("error listening: " + err.Error())
	}
	connErr := make([]error, n.nNodes)
	wg := &sync.WaitGroup{}
	wg.Add(n.nNodes - 1)  
	for i := 0; i < n.nNodes; i++ {
		if i == n.myrank {
			continue
		}
		go func(i int) {
			defer wg.Done()
			connErr[i] = n.listenHandshake(listener)
		}(i)
	}
	wg.Wait()
	var str string
	for _, err := range connErr {
		if err != nil {
			str += " " + err.Error()
		}
	}
	if str != "" {
		return errors.New(str)
	}
	return nil
}
type initialMessage struct {
	Password string  
	Id       int     
}
type listConn struct {
	conn net.Conn
	err  error
}
func (n *Network) listenHandshake(listener net.Listener) error {
	acceptChan := make(chan listConn)
	go func() {
		conn, err := listener.Accept()
		if err != nil {
			err = errors.New("error accepting: " + err.Error())
		}
		acceptChan <- listConn{conn, err}
	}()
	var list listConn
	if n.Timeout > 0 {
		timer := time.NewTimer(n.Timeout)
		select {
		case list = <-acceptChan:
		case <-timer.C:
			list = listConn{
				err: errors.New("listener timed out"),
			}
		}
	} else {
		list = <-acceptChan
	}
	if list.err != nil {
		return list.err
	}
	conn := list.conn
	decoder := gob.NewDecoder(conn)
	var message initialMessage
	err := decoder.Decode(&message)
	if err != nil {
		return err
	}
	id, err := n.passwordAndId(message)
	if err != nil {
		return err
	}
	n.connections[id].listen = conn
	return gob.NewEncoder(conn).Encode(
		initialMessage{
			Password: n.hashedPassword,
			Id:       n.myrank,
		})
}
func (n *Network) establishDialConnections() error {
	connErr := make([]error, n.nNodes)
	wg := &sync.WaitGroup{}
	wg.Add(n.nNodes - 1)  
	for i := 0; i < n.nNodes; i++ {
		if i == n.myrank {
			continue
		}
		go func(i int) {
			defer wg.Done()
			connErr[i] = n.dialHandshake(i)
		}(i)
	}
	wg.Wait()
	var str string
	for _, err := range connErr {
		if err != nil {
			str += " " + err.Error()
		}
	}
	if str != "" {
		return errors.New(str)
	}
	return nil
}
func (n *Network) dialHandshake(i int) error {
	interval := 100 * time.Millisecond
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	var conn net.Conn
	var err error
	start := time.Now()
	for _ = range ticker.C {
		conn, err = net.DialTimeout(n.NetProto, n.Addrs[i], n.Timeout)
		if err == nil || (n.Timeout > 0 && time.Since(start) > n.Timeout) {
			break
		}
	}
	if err != nil {
		return err
	}
	err = gob.NewEncoder(conn).Encode(initialMessage{
		Password: n.hashedPassword,
		Id:       n.myrank,
	})
	if err != nil {
		return err
	}
	var message initialMessage
	err = gob.NewDecoder(conn).Decode(&message)
	if err != nil {
		return err
	}
	id, err := n.passwordAndId(message)
	if err != nil {
		return err
	}
	n.connections[id].dial = conn
	return nil
}
func (n *Network) passwordAndId(message initialMessage) (int, error) {
	if message.Password != n.hashedPassword {
		return -1, errors.New("bad password")
	}
	if message.Id >= n.nNodes || message.Id < 0 || message.Id == n.myrank {
		return -1, fmt.Errorf("bad id: %v", message.Id)
	}
	return message.Id, nil
}
func (n *Network) Finalize() {
	n.close()
}
func (n *Network) close() {
	for _, conn := range n.connections {
		if conn.dial != nil {
			conn.dial.Close()
		}
		if conn.listen != nil {
			conn.listen.Close()
		}
	}
}
type localsend struct {
	c      chan []byte
	create creator
}
type creator int
const (
	notcreated = iota
	sendcreate
	receivecreate
)
type local struct {
	mux   *sync.Mutex
	sends map[int]localsend
}
func newLocal() *local {
	return &local{
		mux:   &sync.Mutex{},
		sends: make(map[int]localsend),
	}
}
func (l *local) Send(tag int, b []byte) {
	l.mux.Lock()
	s, ok := l.sends[tag]
	if ok && s.create == sendcreate {
		panic(fmt.Sprintf("tag %d created twice by sender locally", tag))
	}
	if !ok {
		l.sends[tag] = localsend{
			c:      make(chan []byte),
			create: sendcreate,
		}
	}
	l.mux.Unlock()
	l.sends[tag].c <- b
	l.mux.Lock()
	delete(l.sends, tag)
	l.mux.Unlock()
}
func (l *local) Receive(tag int) []byte {
	l.mux.Lock()
	s, ok := l.sends[tag]
	if ok && s.create == receivecreate {
		panic(fmt.Sprintf("tag %d created twice by receiver locally", tag))
	}
	if !ok {
		l.sends[tag] = localsend{
			c:      make(chan []byte),
			create: receivecreate,
		}
	}
	l.mux.Unlock()
	return <-l.sends[tag].c
}
type tagManager struct {
	CommMap map[int]chan []byte
	Mux     *sync.Mutex
}
func newTagManager() *tagManager {
	return &tagManager{
		CommMap: make(map[int]chan []byte),
		Mux:     &sync.Mutex{},
	}
}
func (t *tagManager) Register(tag int) {
	t.Mux.Lock()
	defer t.Mux.Unlock()
	_, ok := t.CommMap[tag]
	if ok {
		panic(fmt.Sprintf("tag %d already exists on the connection", tag))
	}
	t.CommMap[tag] = make(chan []byte)
}
func (t *tagManager) Delete(tag int) {
	t.Mux.Lock()
	defer t.Mux.Unlock()
	_, ok := t.CommMap[tag]
	if !ok {
		panic("attempt to delete non-existant key")
	}
	delete(t.CommMap, tag)
}