// No need to return NetnsCloser here it will be closed automatically on GC.
// (it may be used in parallel by other parts of the code, so we want it to be used only when not used)
func (p *podNetnsCache) Take(uid string) Netns {
	// lock current pod map
	p.mu.Lock()
	defer p.mu.Unlock()
	if ns, ok := p.currentPodCache[uid]; ok {
		delete(p.currentPodCache, uid)
		// already in cache
		return ns.Netns
	}

	}

	if c.Endpoint == "" {
		// Endpoint not set, make this a no-op
		return nil, nil
	}

	buf, err := r.MarshalMsg(nil)
	if err != nil {
		return nil, err
	}

	// We do not want Gets to take so much time, anything
	// beyond 250ms we should cut it, remote cache is too
	// busy already.
	ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
	defer cancel()

	// If the directory does not exist, it will be created.
	Directory string

	// MaximumFileSize defines the maximum size of each log file in bytes.
	MaximumFileSize int64

	// FileNameFunc specifies the name a new file will take.
	// FileNameFunc must ensure collisions in filenames do not occur.
	// Do not rely on timestamps to be unique, high throughput writes
	// may fall on the same timestamp.
	// Eg.
	// 	2020-03-28_15-00-945-<random-hash>.log

	if workerCount == 0 {
		workerCount = runtime.GOMAXPROCS(0)
	}
	wk, err := workers.New(workerCount)
	if err != nil {
		panic(err)
	}
	for i := 0; i < workerCount; i++ {
		wk.Take()
		go func() {
			defer wk.Give()

			for {
				select {
				case av := <-list.queue:
					list.sendSync(av.ev, av.targetSet)
				case <-list.ctx.Done():
					return
				}
			}
		}()

		// will keep performing the license update. If the leader goes down for some
		// reason, the lock will be released and another node will acquire it and
		// take over because of this loop.
		for {
			licenceUpdaterLoop(ctx, objAPI)

			// license update stopped for some reason.
			// sleep for some time and try again.
			duration := time.Duration(r.Float64() * float64(time.Hour))

	setEventStreamHeaders(w)

	// Listen Publisher and peer-listen-client uses nonblocking send and hence does not wait for slow receivers.
	// Use buffered channel to take care of burst sends or slow w.Write()
	mergeCh := make(chan []byte, globalAPIConfig.getRequestsPoolCapacity()*len(globalEndpoints.Hostnames()))
	localCh := make(chan event.Event, globalAPIConfig.getRequestsPoolCapacity())

		f:      formatBZ2,
	},
}

type untarOptions struct {
	ignoreDirs bool
	ignoreErrs bool
	prefixAll  string
}

// disconnectReader will ensure that no reads can take place on
// the upstream reader after close has been called.
type disconnectReader struct {
	r  io.Reader
	mu sync.Mutex
}

func (d *disconnectReader) Read(p []byte) (n int, err error) {
	d.mu.Lock()

	return int(atomic.LoadInt32(&srv.requestCount))
}

// Init - init HTTP server
func (srv *Server) Init(listenCtx context.Context, listenErrCallback func(listenAddr string, err error)) (serve func() error, err error) {
	// Take a copy of server fields.
	var tlsConfig *tls.Config
	if srv.TLSConfig != nil {
		tlsConfig = srv.TLSConfig.Clone()
	}
	handler := srv.Handler // if srv.Handler holds non-synced state -> possible data race

	fd    uintptr
	inode uint64
}

func (n *NetnsWithFd) Close() error {
	if n.netns == nil {
		return nil
	}

	ret := n.netns.Close()
	// set fd to invalid value, so if something uses it by mistake it will err
	uintZero := uintptr(0)
	n.fd = ^uintZero
	return ret
}

func (n *NetnsWithFd) Fd() uintptr {
	return n.fd
}

func (n *NetnsWithFd) Inode() uint64 {
	return n.inode

		Workload: podToWorkload(pod),
		Netns:    ns,
	}
	netns1 := p.UpsertPodCacheWithNetns(string(pod.UID), wl)
	netnsTaken := p.Take(string(pod.UID))
	if netns1 != netnsTaken {
		t.Fatalf("Expected the original Netns for the same uid, got %p and %p", netns1, ns)
	}
	if nil != p.Take(string(pod.UID)) {
		// expect nil because we already took it
		t.Fatalf("Expected nil Netns for the same uid twice")
	}
}