func (w *WaterMark) Init(closer *Closer) {
	w.markCh = make(chan mark, 100)
	w.elog = trace.NewEventLog("Watermark", w.Name)
	go w.process(closer)
}
func (w *WaterMark) Init(closer *Closer) {
	w.markCh = make(chan mark, 100)
	w.elog = trace.NewEventLog("Watermark", w.Name)
	go w.process(closer)
}
// Init initializes a WaterMark struct. MUST be called before using it.