func main() {

	f, err := os.OpenFile(logFile, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
	if err != nil {
		log.Fatalf("error opening file: %v", err)
	}
	log.SetOutput(io.MultiWriter(f, os.Stdout))

	http.HandleFunc("/", fronthandler)
	http.HandleFunc("/_ah/health", healthhandler)
	go http.ListenAndServe(httpport, nil)

	ce, err := credentials.NewServerTLSFromFile("server_crt.pem", "server_key.pem")
	if err != nil {
		log.Fatalf("Failed to generate credentials %v", err)
	}

	lis, err := net.Listen("tcp", grpcport)
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}

	sopts := []grpc.ServerOption{grpc.MaxConcurrentStreams(10)}
	sopts = append(sopts, grpc.Creds(ce))
	s := grpc.NewServer(sopts...)

	echo.RegisterEchoServerServer(s, &server{})
	log.Printf("Starting gRPC server on port %v", grpcport)

	s.Serve(lis)
}

// Serve serves stuff.
func Serve(
	registerFunc func(*grpc.Server),
	options ServeOptions,
	serveEnv ServeEnv,
) (retErr error) {
	defer func(start time.Time) { logServerFinished(start, retErr) }(time.Now())
	if registerFunc == nil {
		return ErrMustSpecifyRegisterFunc
	}
	if serveEnv.GRPCPort == 0 {
		serveEnv.GRPCPort = 7070
	}
	grpcServer := grpc.NewServer(
		grpc.MaxConcurrentStreams(math.MaxUint32),
		grpc.UnaryInterceptor(protorpclog.LoggingUnaryServerInterceptor),
	)
	registerFunc(grpcServer)
	if options.Version != nil {
		protoversion.RegisterAPIServer(grpcServer, protoversion.NewAPIServer(options.Version, protoversion.APIServerOptions{}))
	}
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", serveEnv.GRPCPort))
	if err != nil {
		return err
	}
	errC := make(chan error)
	go func() { errC <- grpcServer.Serve(listener) }()
	protolion.Info(
		&ServerStarted{
			Port: uint32(serveEnv.GRPCPort),
		},
	)
	return <-errC
}

// NewGrpcServer -
func NewGrpcServer(serv services.Servicer, opts options.Options, logger *logging.Entry) (Serverer, error) {

	addr, addrOk := opts.Get("grpc-addr")

	if !addrOk {
		return nil, errors.New("You must provide `grpc-addr` in order to create gRPC server...")
	}

	listener, err := net.Listen("tcp", addr.String())

	if err != nil {
		return nil, fmt.Errorf("Failed to listen: %v", err)
	}

	var grpcOpts []grpc.ServerOption

	if useGrpc, ok := opts.Get("grpc-tls"); ok && useGrpc.Bool() {
		certFile, certOk := opts.Get("grpc-tls-cert")
		certKey, keyOk := opts.Get("grpc-tls-key")

		if !certOk {
			return nil, errors.New("You must provide `grpc-tls-cert` in order to create gRPC server...")
		}

		if !keyOk {
			return nil, errors.New("You must provide `grpc-tls-key` in order to create gRPC server...")
		}

		creds, err := credentials.NewServerTLSFromFile(certFile.String(), certKey.String())

		if err != nil {
			return nil, fmt.Errorf("Failed to generate gRPC credentials: %v", err)
		}

		grpcOpts = []grpc.ServerOption{grpc.Creds(creds)}
	}

	if maxStreams, msOk := opts.Get("grpc-max-concurrent-streams"); msOk {
		grpcOpts = []grpc.ServerOption{grpc.MaxConcurrentStreams(maxStreams.UInt32())}
	}

	grpcServer := grpc.NewServer(grpcOpts...)

	s := &Grpc{
		Options:           opts,
		Server:            grpcServer,
		Listener:          listener,
		Entry:             logger,
		Servicer:          serv,
		ConnectivityState: &ConnectivityState{},
		Interrupt:         serv.GetInterruptChan(),
	}

	if listenForever, lfOk := opts.Get("grpc-listen-forever"); lfOk {
		s.ListenForever = listenForever.Bool()
	}

	return Serverer(s), nil
}

func do(appEnv *appEnv) error {
	server := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
	protoeasy.RegisterAPIServer(server, protoeasy.DefaultAPIServer)
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", appEnv.Port))
	if err != nil {
		return err
	}
	return server.Serve(listener)
}

func do(appEnvObj interface{}) error {
	appEnv := appEnvObj.(*appEnv)
	pkglog.SetupLogging("protoeasyd", appEnv.LogEnv)

	server := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
	protoeasy.RegisterAPIServer(server, protoeasy.DefaultAPIServer)
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", appEnv.Port))
	if err != nil {
		return err
	}
	return server.Serve(listener)
}

// StartServer starts a gRPC server serving a benchmark service. It returns its
// listen address and a function to stop the server.
func StartServer() (string, func()) {
	lis, err := net.Listen("tcp", ":0")
	if err != nil {
		grpclog.Fatalf("Failed to listen: %v", err)
	}
	s := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
	testpb.RegisterTestServiceServer(s, &testServer{})
	go s.Serve(lis)
	return lis.Addr().String(), func() {
		s.Stop()
	}
}

func do() error {
	runtime.GOMAXPROCS(runtime.NumCPU())

	appEnv := &appEnv{}
	if err := env.Populate(appEnv, env.PopulateOptions{}); err != nil {
		return err
	}
	if appEnv.NumShards == 0 {
		appEnv.NumShards = defaultNumShards
	}
	var btrfsAPI btrfs.API
	switch appEnv.BtrfsDriverType {
	case "exec":
		btrfsAPI = btrfs.NewExecAPI()
	case "ffi":
		fallthrough
	default:
		btrfsAPI = btrfs.NewFFIAPI()
	}
	address := fmt.Sprintf("0.0.0.0:%d", appEnv.APIPort)
	combinedAPIServer := server.NewCombinedAPIServer(
		route.NewSharder(
			appEnv.NumShards,
		),
		route.NewRouter(
			route.NewSingleAddresser(
				address,
				appEnv.NumShards,
			),
			route.NewDialer(),
			address,
		),
		drive.NewBtrfsDriver(
			appEnv.BtrfsRoot,
			btrfsAPI,
		),
	)
	server := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
	pfs.RegisterApiServer(server, combinedAPIServer)
	pfs.RegisterInternalApiServer(server, combinedAPIServer)
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", appEnv.APIPort))
	if err != nil {
		return err
	}

	errC := make(chan error)
	go func() { errC <- server.Serve(listener) }()
	go func() { errC <- http.ListenAndServe(":8080", nil) }()
	if appEnv.TracePort != 0 {
		go func() { errC <- http.ListenAndServe(fmt.Sprintf(":%d", appEnv.TracePort), nil) }()
	}
	return <-errC
}

// StartFlexVolumeAPI starts the flexvolume API on the given port.
func StartFlexVolumeAPI(port uint16) error {
	grpcServer := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
	flexvolume.RegisterAPIServer(grpcServer, flexvolume.NewAPIServer(newFlexVolumeClient()))
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
	if err != nil {
		return err
	}
	go func() {
		if err := grpcServer.Serve(listener); err != nil {
			dlog.Errorln(err.Error())
		}
	}()
	return nil
}

func setUp(hs *health.HealthServer, maxStream uint32, ua string, e env) (s *grpc.Server, cc *grpc.ClientConn) {
	sopts := []grpc.ServerOption{grpc.MaxConcurrentStreams(maxStream)}
	la := ":0"
	switch e.network {
	case "unix":
		la = "/tmp/testsock" + fmt.Sprintf("%d", time.Now())
		syscall.Unlink(la)
	}
	lis, err := net.Listen(e.network, la)
	if err != nil {
		grpclog.Fatalf("Failed to listen: %v", err)
	}
	if e.security == "tls" {
		creds, err := credentials.NewServerTLSFromFile(tlsDir+"server1.pem", tlsDir+"server1.key")
		if err != nil {
			grpclog.Fatalf("Failed to generate credentials %v", err)
		}
		sopts = append(sopts, grpc.Creds(creds))
	}
	s = grpc.NewServer(sopts...)
	if hs != nil {
		healthpb.RegisterHealthCheckServer(s, hs)
	}
	testpb.RegisterTestServiceServer(s, &testServer{})
	go s.Serve(lis)
	addr := la
	switch e.network {
	case "unix":
	default:
		_, port, err := net.SplitHostPort(lis.Addr().String())
		if err != nil {
			grpclog.Fatalf("Failed to parse listener address: %v", err)
		}
		addr = "localhost:" + port
	}
	if e.security == "tls" {
		creds, err := credentials.NewClientTLSFromFile(tlsDir+"ca.pem", "x.test.youtube.com")
		if err != nil {
			grpclog.Fatalf("Failed to create credentials %v", err)
		}
		cc, err = grpc.Dial(addr, grpc.WithTransportCredentials(creds), grpc.WithDialer(e.dialer), grpc.WithUserAgent(ua))
	} else {
		cc, err = grpc.Dial(addr, grpc.WithDialer(e.dialer), grpc.WithUserAgent(ua))
	}
	if err != nil {
		grpclog.Fatalf("Dial(%q) = %v", addr, err)
	}
	return
}

func setUp(maxStream uint32, e env) (s *grpc.Server, cc *grpc.ClientConn) {
	s = grpc.NewServer(grpc.MaxConcurrentStreams(maxStream))
	la := ":0"
	switch e.network {
	case "unix":
		la = "/tmp/testsock" + fmt.Sprintf("%p", s)
		syscall.Unlink(la)
	}
	lis, err := net.Listen(e.network, la)
	if err != nil {
		log.Fatalf("Failed to listen: %v", err)
	}
	testpb.RegisterTestServiceServer(s, &testServer{})
	if e.security == "tls" {
		creds, err := credentials.NewServerTLSFromFile(tlsDir+"server1.pem", tlsDir+"server1.key")
		if err != nil {
			log.Fatalf("Failed to generate credentials %v", err)
		}
		go s.Serve(creds.NewListener(lis))
	} else {
		go s.Serve(lis)
	}
	addr := la
	switch e.network {
	case "unix":
	default:
		_, port, err := net.SplitHostPort(lis.Addr().String())
		if err != nil {
			log.Fatalf("Failed to parse listener address: %v", err)
		}
		addr = "localhost:" + port
	}
	if e.security == "tls" {
		creds, err := credentials.NewClientTLSFromFile(tlsDir+"ca.pem", "x.test.youtube.com")
		if err != nil {
			log.Fatalf("Failed to create credentials %v", err)
		}
		cc, err = grpc.Dial(addr, grpc.WithTransportCredentials(creds), grpc.WithDialer(e.dialer))
	} else {
		cc, err = grpc.Dial(addr, grpc.WithDialer(e.dialer))
	}
	if err != nil {
		log.Fatalf("Dial(%q) = %v", addr, err)
	}
	return
}

func GrpcDo(
	port int,
	tracePort int,
	version *protoversion.Version,
	registerFunc func(*grpc.Server),
) error {
	s := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
	registerFunc(s)
	protoversion.RegisterApiServer(s, protoversion.NewAPIServer(version))
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
	if err != nil {
		return err
	}
	errC := make(chan error)
	go func() { errC <- s.Serve(listener) }()
	if tracePort != 0 {
		go func() { errC <- http.ListenAndServe(fmt.Sprintf(":%d", tracePort), nil) }()
	}
	return <-errC
}

func (g *grpcSuite) SetupSuite() {
	g.servers = make(map[string]*grpc.Server)
	listeners := make(map[string]net.Listener)
	ports, err := getPorts(g.numServers)
	require.NoError(g.T(), err)
	for i := 0; i < g.numServers; i++ {
		port := ports[i]
		require.NoError(g.T(), err)
		address := fmt.Sprintf("0.0.0.0:%s", port)
		server := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
		g.servers[address] = server
		listener, err := net.Listen("tcp", fmt.Sprintf(":%s", port))
		require.NoError(g.T(), err)
		listeners[address] = listener
	}
	g.registerFunc(g.servers)
	g.errC = make(chan error, g.numServers)
	for address, server := range g.servers {
		address := address
		server := server
		go func() {
			g.errC <- server.Serve(listeners[address])
		}()
	}
	g.done = make(chan bool, 1)
	go func() {
		for j := 0; j < g.numServers; j++ {
			<-g.errC
		}
		g.done <- true
	}()
	g.clientConns = make(map[string]*grpc.ClientConn)
	for address := range g.servers {
		clientConn, err := grpc.Dial(address)
		if err != nil {
			g.TearDownSuite()
			require.NoError(g.T(), err)
		}
		g.clientConns[address] = clientConn
	}
}

// NewServer creates a qrpc server which has not started to accept requests yet.
func NewServer(cfg *Config) *Server {
	log.Printf("-> NewServer(%v)\n", cfg)

	return &Server{
		cfg: cfg,
		rpc: grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32)),
		cluster: &cluster{
			key:     fmt.Sprintf("peer-%s", keySuffix()),
			timeout: cfg.ClusterRequestTimeout,
			peers:   make(map[string]*internal.Peer),
		},
		cancelWatcher: nil,

		mtx: &sync.Mutex{},
		data: diskv.New(diskv.Options{
			BasePath:     cfg.DataBasePath,
			Transform:    transformKey,
			CacheSizeMax: cfg.MaxCacheSize,
		}),
	}
}

func serverSetUp(t *testing.T, servON bool, hs *health.HealthServer, maxStream uint32, cp grpc.Compressor, dc grpc.Decompressor, e env) (s *grpc.Server, addr string) {
	t.Logf("Running test in %s environment...", e.name)
	sopts := []grpc.ServerOption{grpc.MaxConcurrentStreams(maxStream), grpc.RPCCompressor(cp), grpc.RPCDecompressor(dc)}
	la := ":0"
	switch e.network {
	case "unix":
		la = "/tmp/testsock" + fmt.Sprintf("%d", time.Now())
		syscall.Unlink(la)
	}
	lis, err := net.Listen(e.network, la)
	if err != nil {
		t.Fatalf("Failed to listen: %v", err)
	}
	if e.security == "tls" {
		creds, err := credentials.NewServerTLSFromFile(tlsDir+"server1.pem", tlsDir+"server1.key")
		if err != nil {
			t.Fatalf("Failed to generate credentials %v", err)
		}
		sopts = append(sopts, grpc.Creds(creds))
	}
	s = grpc.NewServer(sopts...)
	if hs != nil {
		healthpb.RegisterHealthServer(s, hs)
	}
	if servON {
		testpb.RegisterTestServiceServer(s, &testServer{security: e.security})
	}
	go s.Serve(lis)
	addr = la
	switch e.network {
	case "unix":
	default:
		_, port, err := net.SplitHostPort(lis.Addr().String())
		if err != nil {
			t.Fatalf("Failed to parse listener address: %v", err)
		}
		addr = "localhost:" + port
	}
	return
}

// Serve serves stuff.
func Serve(
	registerFunc func(*grpc.Server),
	options ServeOptions,
	serveEnv ServeEnv,
) (retErr error) {
	defer func(start time.Time) { logServerFinished(start, retErr) }(time.Now())
	if registerFunc == nil {
		return ErrMustSpecifyRegisterFunc
	}
	if serveEnv.GRPCPort == 0 {
		serveEnv.GRPCPort = 7070
	}
	grpcServer := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
	registerFunc(grpcServer)
	if options.Version != nil {
		protoversion.RegisterAPIServer(grpcServer, protoversion.NewAPIServer(options.Version, protoversion.APIServerOptions{}))
	}
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", serveEnv.GRPCPort))
	if err != nil {
		return err
	}
	return grpcServer.Serve(listener)
}

func do() error {
	runtime.GOMAXPROCS(runtime.NumCPU())

	appEnv := &appEnv{}
	if err := env.Populate(appEnv, env.PopulateOptions{}); err != nil {
		return err
	}
	//address := fmt.Sprintf("0.0.0.0:%d", appEnv.APIPort)
	s := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
	pps.RegisterApiServer(s, server.NewAPIServer(store.NewInMemoryClient()))
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", appEnv.APIPort))
	if err != nil {
		return err
	}

	errC := make(chan error)
	go func() { errC <- s.Serve(listener) }()
	//go func() { errC <- http.ListenAndServe(":8080", nil) }()
	if appEnv.TracePort != 0 {
		go func() { errC <- http.ListenAndServe(fmt.Sprintf(":%d", appEnv.TracePort), nil) }()
	}
	return <-errC
}

func setUp(useTLS bool, maxStream uint32) (s *grpc.Server, tc testpb.TestServiceClient) {
	lis, err := net.Listen("tcp", ":0")
	if err != nil {
		log.Fatalf("Failed to listen: %v", err)
	}
	_, port, err := net.SplitHostPort(lis.Addr().String())
	if err != nil {
		log.Fatalf("Failed to parse listener address: %v", err)
	}
	s = grpc.NewServer(grpc.MaxConcurrentStreams(maxStream))
	testpb.RegisterTestServiceServer(s, &testServer{})
	if useTLS {
		creds, err := credentials.NewServerTLSFromFile(tlsDir+"server1.pem", tlsDir+"server1.key")
		if err != nil {
			log.Fatalf("Failed to generate credentials %v", err)
		}
		go s.Serve(creds.NewListener(lis))
	} else {
		go s.Serve(lis)
	}
	addr := "localhost:" + port
	var conn *grpc.ClientConn
	if useTLS {
		creds, err := credentials.NewClientTLSFromFile(tlsDir+"ca.pem", "x.test.youtube.com")
		if err != nil {
			log.Fatalf("Failed to create credentials %v", err)
		}
		conn, err = grpc.Dial(addr, grpc.WithTransportCredentials(creds))
	} else {
		conn, err = grpc.Dial(addr)
	}
	if err != nil {
		log.Fatalf("Dial(%q) = %v", addr, err)
	}
	tc = testpb.NewTestServiceClient(conn)
	return
}

// ServeWithHTTP serves stuff.
func ServeWithHTTP(
	registerFunc func(*grpc.Server),
	httpRegisterFunc func(context.Context, *runtime.ServeMux, *grpc.ClientConn) error,
	options ServeWithHTTPOptions,
	serveEnv ServeEnv,
	handlerEnv pkghttp.HandlerEnv,
) (retErr error) {
	defer func(start time.Time) { logServerFinished(start, retErr) }(time.Now())
	if registerFunc == nil || httpRegisterFunc == nil {
		return ErrMustSpecifyRegisterFunc
	}
	if serveEnv.GRPCPort == 0 {
		serveEnv.GRPCPort = 7070
	}
	if handlerEnv.Port == 0 {
		handlerEnv.Port = 8080
	}

	grpcServer := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
	registerFunc(grpcServer)
	if options.Version != nil {
		protoversion.RegisterAPIServer(grpcServer, protoversion.NewAPIServer(options.Version, protoversion.APIServerOptions{}))
	}
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", serveEnv.GRPCPort))
	if err != nil {
		return err
	}
	grpcErrC := make(chan error)
	go func() { grpcErrC <- grpcServer.Serve(listener) }()

	time.Sleep(1 * time.Second)
	ctx, cancel := context.WithCancel(context.Background())
	conn, err := grpc.Dial(fmt.Sprintf("0.0.0.0:%d", serveEnv.GRPCPort), grpc.WithInsecure())
	if err != nil {
		cancel()
		return err
	}
	go func() {
		<-ctx.Done()
		_ = conn.Close()
	}()
	mux := runtime.NewServeMux()
	if options.Version != nil {
		if err := protoversion.RegisterAPIHandler(ctx, mux, conn); err != nil {
			cancel()
			return err
		}
	}
	if err := httpRegisterFunc(ctx, mux, conn); err != nil {
		cancel()
		return err
	}
	var handler http.Handler = mux
	if options.HTTPHandlerModifier != nil {
		handler, err = options.HTTPHandlerModifier(mux)
		if err != nil {
			cancel()
			return err
		}
	}
	httpErrC := make(chan error)
	go func() { httpErrC <- pkghttp.ListenAndServe(handler, handlerEnv) }()
	protolion.Info(
		&ServerStarted{
			Port:     uint32(serveEnv.GRPCPort),
			HttpPort: uint32(handlerEnv.Port),
		},
	)
	var errs []error
	grpcStopped := false
	for i := 0; i < 2; i++ {
		select {
		case grpcErr := <-grpcErrC:
			if grpcErr != nil {
				errs = append(errs, fmt.Errorf("grpc error: %s", grpcErr.Error()))
			}
			grpcStopped = true
		case httpErr := <-httpErrC:
			if httpErr != nil {
				errs = append(errs, fmt.Errorf("http error: %s", httpErr.Error()))
			}
			if !grpcStopped {
				grpcServer.Stop()
				_ = listener.Close()
				grpcStopped = true
			}
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("%v", errs)
	}
	return nil
}

// Serve serves stuff.
func Serve(
	port uint16,
	registerFunc func(*grpc.Server),
	opts ServeOptions,
) (retErr error) {
	start := time.Now()
	defer func() {
		if retErr != nil {
			protolog.Error(
				&ServerFinished{
					Error:    retErr.Error(),
					Duration: prototime.DurationToProto(time.Since(start)),
				},
			)
		} else {
			protolog.Info(
				&ServerFinished{
					Duration: prototime.DurationToProto(time.Since(start)),
				},
			)
		}
	}()
	if port == 0 {
		return ErrMustSpecifyPort
	}
	if registerFunc == nil {
		return ErrMustSpecifyRegisterFunc
	}
	if opts.HTTPPort != 0 && opts.HTTPAddress != "" {
		return ErrCannotSpecifyBothHTTPPortAndHTTPAddress
	}
	s := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
	registerFunc(s)
	if opts.Version != nil {
		protoversion.RegisterAPIServer(s, protoversion.NewAPIServer(opts.Version))
	}
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
	if err != nil {
		return err
	}
	grpcErrC := make(chan error)
	grpcDebugErrC := make(chan error)
	httpErrC := make(chan error)
	errCCount := 1
	go func() { grpcErrC <- s.Serve(listener) }()
	if opts.DebugPort != 0 {
		errCCount++
		debugServer := &graceful.Server{
			Timeout: 1 * time.Second,
			Server: &http.Server{
				Addr:    fmt.Sprintf(":%d", opts.DebugPort),
				Handler: http.DefaultServeMux,
			},
		}
		go func() { grpcDebugErrC <- debugServer.ListenAndServe() }()
	}
	if (opts.HTTPPort != 0 || opts.HTTPAddress != "") && (opts.Version != nil || opts.HTTPRegisterFunc != nil) {
		time.Sleep(1 * time.Second)
		ctx, cancel := context.WithCancel(context.Background())
		var mux *runtime.ServeMux
		if len(opts.ServeMuxOptions) == 0 {
			mux = runtime.NewServeMux()
		} else {
			mux = runtime.NewServeMux(opts.ServeMuxOptions...)
		}
		conn, err := grpc.Dial(fmt.Sprintf("0.0.0.0:%d", port), grpc.WithInsecure())
		if err != nil {
			glog.Flush()
			cancel()
			return err
		}
		go func() {
			<-ctx.Done()
			_ = conn.Close()
		}()
		if opts.Version != nil {
			if err := protoversion.RegisterAPIHandler(ctx, mux, conn); err != nil {
				_ = conn.Close()
				glog.Flush()
				cancel()
				return err
			}
		}
		if opts.HTTPRegisterFunc != nil {
			if err := opts.HTTPRegisterFunc(ctx, mux, conn); err != nil {
				_ = conn.Close()
				glog.Flush()
				cancel()
				return err
			}
		}
		httpAddress := fmt.Sprintf(":%d", opts.HTTPPort)
		if opts.HTTPAddress != "" {
			httpAddress = opts.HTTPAddress
		}
		httpServer := &http.Server{
			Addr:    httpAddress,
			Handler: mux,
		}
//.........这里部分代码省略.........

func (s *server) Serve() (retErr error) {
	var listener net.Listener
	var spec string
	var err error
	var addr string
	switch s.protocol {
	case protocolTCP:
		listener, spec, err = newTCPListener(s.driverName, s.groupOrAddress)
		addr = s.groupOrAddress
	case protocolUnix:
		listener, spec, err = newUnixListener(s.driverName, s.groupOrAddress)
		addr = s.driverName
	default:
		return fmt.Errorf("unknown protocol: %d", s.protocol)
	}
	if err != nil {
		return err
	}
	grpcPort := s.opts.GRPCPort
	if grpcPort == 0 {
		grpcPort = DefaultGRPCPort
	}

	grpcServer := grpc.NewServer(grpc.MaxConcurrentStreams(math.MaxUint32))
	RegisterAPIServer(grpcServer, s.apiServer)
	s.registerFunc(grpcServer)
	grpcListener, err := net.Listen("tcp", fmt.Sprintf(":%d", grpcPort))
	if err != nil {
		return err
	}
	grpcErrC := make(chan error)
	httpErrC := make(chan error)
	errCCount := 1
	go func() { grpcErrC <- grpcServer.Serve(grpcListener) }()
	time.Sleep(1 * time.Second)
	ctx, cancel := context.WithCancel(context.Background())
	mux := runtime.NewServeMux(
		runtime.WithForwardResponseOption(
			func(_ context.Context, responseWriter http.ResponseWriter, _ proto.Message) error {
				responseWriter.Header().Set("Content-Type", "application/vnd.docker.plugins.v1.1+json")
				return nil
			},
		),
	)
	conn, err := grpc.Dial(fmt.Sprintf("0.0.0.0:%d", grpcPort), grpc.WithInsecure())
	if err != nil {
		glog.Flush()
		cancel()
		return err
	}
	go func() {
		<-ctx.Done()
		_ = conn.Close()
	}()
	if err := RegisterAPIHandler(ctx, mux, conn); err != nil {
		_ = conn.Close()
		glog.Flush()
		cancel()
		return err
	}
	if err := s.httpRegisterFunc(ctx, mux, conn); err != nil {
		_ = conn.Close()
		glog.Flush()
		cancel()
		return err
	}
	httpServer := &http.Server{
		Addr:    addr,
		Handler: mux,
	}
	gracefulServer := &graceful.Server{
		Timeout: 1 * time.Second,
		ShutdownInitiated: func() {
			glog.Flush()
			cancel()
			if spec != "" {
				_ = os.Remove(spec)
			}
		},
		Server: httpServer,
	}
	errCCount++
	go func() {
		httpErrC <- gracefulServer.Serve(listener)
	}()
	var errs []error
	grpcStopped := false
	for i := 0; i < errCCount; i++ {
		select {
		case grpcErr := <-grpcErrC:
			if grpcErr != nil {
				errs = append(errs, fmt.Errorf("grpc error: %s", grpcErr.Error()))
			}
			grpcStopped = true
		case httpErr := <-httpErrC:
			if httpErr != nil {
				errs = append(errs, fmt.Errorf("http error: %s", httpErr.Error()))
			}
			if !grpcStopped {
				grpcServer.Stop()
//.........这里部分代码省略.........