func (s *Supervisor) wsLog(w http.ResponseWriter, r *http.Request) {
	name := mux.Vars(r)["name"]
	log.Println(name)
	proc, ok := s.procMap[name]
	if !ok {
		log.Println("No such process")
		// TODO: raise error here?
		return
	}

	c, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Print("upgrade:", err)
		return
	}
	defer c.Close()

	for data := range proc.Output.NewChanString(r.RemoteAddr) {
		err := c.WriteMessage(1, []byte(data))
		if err != nil {
			proc.Output.CloseWriter(r.RemoteAddr)
			break
		}
	}
}

func (s *Supervisor) wsEvents(w http.ResponseWriter, r *http.Request) {
	c, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Print("upgrade:", err)
		return
	}
	defer c.Close()

	ch := make(chan string, 0)
	s.addStatusChangeListener(ch)
	go func() {
		_, _ = <-ch // ignore the history messages
		for message := range ch {
			// Question: type 1 ?
			c.WriteMessage(1, []byte(message))
		}
		// s.eventB.RemoveListener(ch)
	}()
	for {
		mt, message, err := c.ReadMessage()
		if err != nil {
			log.Println("read:", mt, err)
			break
		}
		log.Printf("recv: %v %s", mt, message)
		err = c.WriteMessage(mt, message)
		if err != nil {
			log.Println("write:", err)
			break
		}
	}
}

func startWorker() {
	log.Println("start Worker")
	Cronjob.Clear()
	Cronjob.AddFunc(Config.Flow.Cron,  func() { log.Println(Config.Flow.Cron + ": start DumpFlow()"); go DumpFlow(); })
	//Cronjob.AddFunc("@every 1s", func() { fmt.Println("Every 1") })
	Cronjob.Start()
	//defer Cronjob.Stop()
	//Cronjob.Entries()
}

func (s *Supervisor) addOrUpdateProgram(pg Program) error {
	// defer s.broadcastEvent(pg.Name + " add or update")
	if err := pg.Check(); err != nil {
		return err
	}
	origPg, ok := s.pgMap[pg.Name]
	if ok {
		if reflect.DeepEqual(origPg, pg) {
			return nil
		}
		s.broadcastEvent(pg.Name + " update")
		log.Println("Update:", pg.Name)
		origProc := s.procMap[pg.Name]
		isRunning := origProc.IsRunning()
		go func() {
			s.stopAndWait(origProc.Name)

			newProc := s.newProcess(pg)
			s.procMap[pg.Name] = newProc
			s.pgMap[pg.Name] = pg // update origin
			if isRunning {
				newProc.Operate(StartEvent)
			}
		}()
	} else {
		s.names = append(s.names, pg.Name)
		s.pgMap[pg.Name] = pg
		s.procMap[pg.Name] = s.newProcess(pg)
		s.broadcastEvent(pg.Name + " added")
	}
	return nil
}

func actionConfigTest(c *cli.Context) error {
	if _, _, err := newSupervisorHandler(); err != nil {
		log.Fatal(err)
	}
	log.Println("test is successful")
	return nil
}

func (p *Process) startCommand() {
	// p.Stdout.Reset()
	// p.Stderr.Reset()
	// p.Output.Reset() // Donot reset because log is still needed.
	log.Printf("start cmd(%s): %s", p.Name, p.Command)
	p.cmd = p.buildCommand()

	p.SetState(Running)
	if err := p.cmd.Start(); err != nil {
		log.Warnf("program %s start failed: %v", p.Name, err)
		p.SetState(Fatal)
		return
	}
	go func() {
		errC := GoFunc(p.cmd.Wait)
		startTime := time.Now()
		select {
		case <-errC:
			// if p.cmd.Wait() returns, it means program and its sub process all quited. no need to kill again
			// func Wait() will only return when program session finishs. (Only Tested on mac)
			log.Printf("program(%s) finished, time used %v", p.Name, time.Since(startTime))
			if time.Since(startTime) < time.Duration(p.StartSeconds)*time.Second {
				if p.retryLeft == p.StartRetries { // If first time quit so fast, just set to fatal
					p.SetState(Fatal)
					log.Printf("program(%s) exit too quick, status -> fatal", p.Name)
					return
				}
			}
			p.waitNextRetry()
		case <-p.stopC:
			log.Println("recv stop command")
			p.stopCommand() // clean up all process
		}
	}()
}

func DownloadImg(url_ string) {
	resp, err := http.Get(url_)
	if goutils.CheckErr(err) {
		return
	}
	log.Println(resp.Header.Get("Content-Type"))
	log.Println(resp.ContentLength)
	_ = resp
	b, err := ioutil.ReadAll(resp.Body)
	if goutils.CheckErr(err) {
		return
	}
	// log.Println(b)
	file, _ := os.OpenFile("down.png", os.O_CREATE|os.O_WRONLY, 0644)
	file.Write(b)
}

func proxyHandler(w http.ResponseWriter, r *http.Request) {
	hj, ok := w.(http.Hijacker)
	if !ok {
		http.Error(w, "webserver don't support hijacking", http.StatusInternalServerError)
		return
	}

	var proxyFor = r.Header.Get("X-Proxy-For")
	log.Println("X-Proxy-For client name:", r.RemoteAddr, proxyFor)

	connCh, ok := namedConnection[proxyFor]
	if !ok {
		http.Error(w, "inside error: proxy not ready to receive conn", http.StatusInternalServerError)
		return
	}
	hjconn, bufrw, err := hj.Hijack()
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		connCh <- nil
		return
	}
	//if _, ok := hjconn.(*net.TCPConn); ok {
	//log.Println("Hijack is tcp conn")
	//}

	conn := NewHijackReadWriteCloser(hjconn.(*net.TCPConn), bufrw)
	connCh <- conn
}

func (k *Keeper) NewRecord(name string) (key string, rec *Record, err error) {
	k.tkmu.Lock()
	defer k.tkmu.Unlock()

	task, ok := k.tasks[name]
	if !ok {
		return "", nil, fmt.Errorf("No such task has name: %s", name)
	}

	xe.ShowSQL = true
	total, err := xe.Where("name = ?", name).Count(&Record{})
	log.Println("CC:", total, err)

	// order update
	k.removeFromOrder(name)
	k.taskOrder = append([]string{name}, k.taskOrder...)

	idx := int(total)
	rec = &Record{
		Name:   name,
		Index:  idx,
		Buffer: bytes.NewBuffer(nil),
		T:      task,
	}
	key = rec.Key()
	k.crmu.Lock()
	k.runRecs[key] = rec
	k.crmu.Unlock()
	return key, rec, nil
}

func (p *ProxyConn) start() {
	defer p.lconn.Close()
	defer p.rconn.Close()
	//connect to remote
	// rconn, err := net.DialTCP("tcp", nil, p.raddr)
	// if err != nil {
	// 	log.Printf("Remote connection failed: %s", err)
	// 	return
	// }
	// p.rconn = rconn
	// defer p.rconn.Close()

	// FIXME: may need to set a flag
	if tcpconn, ok := p.lconn.(*net.TCPConn); ok {
		tcpconn.SetNoDelay(true)
	}
	if tcpconn, ok := p.rconn.(*net.TCPConn); ok {
		tcpconn.SetNoDelay(true)
	}
	// p.lconn.SetNoDelay(true)
	// p.rconn.SetNoDelay(true)

	//display both ends
	// log.Printf("Opened %s >>> %s", p.lconn.RemoteAddr().String(), p.rconn.RemoteAddr().String())
	//bidirectional copy
	wg := sync.WaitGroup{}
	wg.Add(2)
	go func() {
		ch1 := p.pipe(p.lconn, p.rconn)
		<-ch1
		closeRead(p.lconn)
		closeWrite(p.rconn)
		log.Println("close local -> remote")
		wg.Done()
	}()
	go func() {
		ch2 := p.pipe(p.rconn, p.lconn)
		<-ch2
		closeRead(p.rconn)
		closeWrite(p.lconn)
		log.Println("close remote -> local")
		wg.Done()
	}()
	wg.Wait()
	//wait for close...
	// log.Printf("Closed (%d bytes sent, %d bytes recieved)", p.sentBytes, p.receivedBytes)
}

func main() {
	flag.Parse()

	log.SetOutputLevel(*flagdebug)

	configFileName := filepath.Clean(*flagconfig + "/main.conf")
	log.Debugf("config file '%s'\n", configFileName)

	log.Printf("Starting %s\n", VERSION)

	if *cpuprofile != "" {
		prof, err := os.Create(*cpuprofile)
		if err != nil {
			panic(err.Error())
		}

		pprof.StartCPUProfile(prof)
		defer func() {
			log.Println("closing file")
			prof.Close()
		}()
		defer func() {
			log.Println("stopping profile")
			pprof.StopCPUProfile()
		}()
	}

	go configWatcher(configFileName)

	terminate := make(chan os.Signal)
	signal.Notify(terminate, os.Interrupt)

	<-terminate
	log.Printf("signal received, stopping")

	if *memprofile != "" {
		f, err := os.Create(*memprofile)
		if err != nil {
			log.Fatal(err)
		}
		pprof.WriteHeapProfile(f)
		f.Close()
	}

	//os.Exit(0)

}

func update_popup() (ReadEntity, error) {
	fs, fes, err := feedentries_updated()
	if err == nil {
		return ReadEntity{ReadSource: fs, Entries: fes}, nil
	}
	log.Println("update-popup", err)
	return ReadEntity{}, err
}

func load_gfwrules() {
	fp := filepath.Join(backend_config().DataFolder, "gfwlist.txt")
	reader, err := os.Open(fp)
	if err == nil {
		defer reader.Close()
		backend_context.ruler, err = gfwlist.NewGfwRuler(reader)
	}
	log.Println("gfwlist-load", err, fp)
}

func actionStatus(c *cli.Context) error {
	err := checkServerStatus()
	if err != nil {
		log.Fatal(err)
	} else {
		log.Println("Server is running, OK.")
	}
	return nil
}

func addRoute(m *macaron.Macaron, method, patten, script, contentType string) {
	log.Println("Add Route:", method, patten, script)
	method = strings.ToUpper(method)
	switch method {
	case "GET":
		m.Get(patten, NewScriptHandler(script, contentType))
	case "POST":
		m.Post(patten, NewScriptHandler(script, contentType))
	}
}

// Performance
func (s *Supervisor) wsPerf(w http.ResponseWriter, r *http.Request) {
	c, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Print("upgrade:", err)
		return
	}
	defer c.Close()

	name := mux.Vars(r)["name"]
	proc, ok := s.procMap[name]
	if !ok {
		log.Println("No such process")
		// TODO: raise error here?
		return
	}
	for {
		// c.SetWriteDeadline(time.Now().Add(3 * time.Second))
		if proc.cmd == nil || proc.cmd.Process == nil {
			log.Println("process not running")
			return
		}
		pid := proc.cmd.Process.Pid
		ps, err := gops.NewProcess(pid)
		if err != nil {
			break
		}
		mainPinfo, err := ps.ProcInfo()
		if err != nil {
			break
		}
		pi := ps.ChildrenProcInfo(true)
		pi.Add(mainPinfo)

		err = c.WriteJSON(pi)
		if err != nil {
			break
		}
		time.Sleep(700 * time.Millisecond)
	}
}

func handleRevConn(pURL *url.URL, lis net.Listener) {
	switch pURL.Scheme {
	case "tcp":
		for {
			rconn, err := lis.Accept()
			if err != nil {
				log.Errorf("accept error: %v", err)
				return
			}
			log.Info("dial local:", pURL)
			lconn, err := net.Dial("tcp", pURL.Host)
			if err != nil {
				// wsclient
				log.Println(err)
				rconn.Close()
				break
			}
			// start forward local proxy
			pc := &ProxyConn{
				lconn: lconn,
				rconn: rconn,
				stats: proxyStats,
			}
			go pc.start()
		}
	case "http", "https":
		remote := pURL
		rp := &httputil.ReverseProxy{
			Director: func(req *http.Request) {
				req.Host = remote.Host
				req.URL.Scheme = remote.Scheme
				req.URL.Host = remote.Host
			},
		}
		http.Serve(lis, rp)
	default:
		log.Println("Unknown protocol:", pURL.Scheme)
	}
}

func feedentries_updated() (ReadSource, []ReadEntry, error) {
	bcms := baidu.NewBcmsProxy(baiduq)
	var v pubsub.PubsubMessage
	err := bcms.FetchOneAsJson(&v)
	if err != nil {
		return ReadSource{}, nil, err
	}
	if (v.Status.StatusCode != 200 && v.Status.StatusCode != 0) || v.Status.Feed == "" {
		return ReadSource{}, nil, new_backenderror(v.Status.StatusCode, v.Status.StatusReason)
	}
	fs := ReadSource{
		FeedSource: feed.FeedSource{
			Name:        v.Title,
			Uri:         v.Status.Feed,
			Description: v.Subtitle,
			Period:      v.Status.Period / 60,
			Update:      int64(v.Updated),
		},
		LastTouch:  int64(unixtime.TimeNow()),
		NextTouch:  int64(unixtime.Time(v.Status.Period) + unixtime.TimeNow()),
		LastUpdate: int64(unixtime.TimeNow()),
	}
	if fs.Period == 0 {
		fs.Period = 120 // minutes
	}
	fes := make([]ReadEntry, len(v.Items))
	for idx, i := range v.Items {
		fes[idx] = ReadEntry{
			FeedEntry: feed.FeedEntry{
				Uri:     i.Uri,
				Title:   i.Title,
				PubDate: int64(i.Published),
				Summary: i.Summary,
				Content: i.Content,
				Tags:    i.Categories,
			},
		}
		feedentry_init_from_standardlinks(i.StandardLinks, fes[idx])
		if fes[idx].Uri == "" {
			feedentry_init_from_links(i.Links, fes[idx])
		}
	}
	if err == nil {
		fes = readentry_filter(fes)
		fo := new_feedsource_operator()
		// ignore touch error, because, source may not be subscribed
		fo.touch(fs.Uri, int64(fs.LastTouch), int64(fs.NextTouch), fs.Period)
	}
	log.Println("updated", fs.Name, fs.Update)
	return fs, fes, err
}

// source : feed(atom/rss) url
// count: must large than 0
// page: 0 based page index
// if page is 0, entries may be fetched online
func feedentry_unread(source string, count int, page int) ([]ReadEntry, error, int) {
	if count <= 0 {
		panic("invalid arg count")
	}
	var sc int
	if page == 0 {
		log.Println("curl-get...")
		c := curl.NewCurlerDetail(backend_config().FeedEntryFolder, 0, 0, nil, backend_context.ruler)
		cache, err := c.GetUtf8(source)
		log.Println("curl-get", cache.LocalUtf8)
		if err != nil || cache.LocalUtf8 == "" {
			return nil, err, cache.StatusCode
		}

		ext := curl.MimeToExt(cache.Mime)
		if ext != "xml" && ext != "atom+xml" && ext != "rss+xml" {
			return nil, new_backenderror(cache.StatusCode, "unsupported mime: "+cache.Mime), 0
		}
		f, err := os.Open(cache.LocalUtf8)
		if err != nil {
			return nil, err, cache.StatusCode
		}

		fs, v, err := feed.NewFeedMaker(f, source).MakeFeed()
		f.Close()
		rs := new_readsource(fs)
		if err == nil {
			new_feedsource_operator().update(rs)
			log.Println("feed-update", fs.Name)
		}
		rv := readentry_filter(new_readentries(v))
		log.Println("feedentries-filter", len(rv))
		sc = cache.StatusCode
	}
	rv, err := new_feedentry_operator().topn_by_feedsource(count*page, count, source)
	log.Println("unread-return(uri, page, count)", source, page, count, len(rv), err)
	return rv, err, sc
}

func update_work() {
	backend_context.Lock()
	defer backend_context.Unlock()
	w := atomic.AddInt64(&backend_context.working, 1)
	defer atomic.AddInt64(&backend_context.working, -1)
	if w != 1 {
		return
	}
	fss, err := feedsource_expired(time.Now().Unix())
	if err != nil || len(fss) == 0 {
		return
	}
	idx := rand.Intn(len(fss))
	fs := fss[idx]
	newfs, fes, err := feed_fetch(fs.Uri)
	newfs.Type = fs.Type
	newfs.EnableProxy = fs.EnableProxy
	newfs.Categories = append(newfs.Categories, fs.Categories...)
	if newfs.Logo == "" {
		newfs.Logo = fs.Logo
	}

	newfs.SubscribeState = fs.SubscribeState
	newfs.LastTouch = int64(unixtime.TimeNow())
	newfs.LastUpdate = newfs.LastTouch
	newfs.NextTouch = newfs.Period + newfs.LastTouch
	err = feedsource_save(newfs)
	fes = readentry_filter(fes)
	backend_push_update(newfs, fes, err)
	ps := pubsub.NewSuperFeedrPubSubscriber("async", "Hearts", "Refresh")
	sc, err := ps.Subscribe(fs.Uri)
	if err != nil {
		log.Println("pubsub-google", sc, err)
	}
	ps = pubsub.NewGooglePubSubscriber()
	sc, err = ps.Subscribe(fs.Uri)
	log.Println("update-tick", fs.Name, sc, err)
}