diff --git a/common/queue/queue.go b/common/queue/queue.go deleted file mode 100644 index 777b63b..0000000 --- a/common/queue/queue.go +++ /dev/null @@ -1,71 +0,0 @@ -package queue - -import ( - "sync" -) - -// Queue is a simple concurrent safe queue -type Queue struct { - items []interface{} - lock sync.RWMutex -} - -// Put add the item to the queue. -func (q *Queue) Put(items ...interface{}) { - if len(items) == 0 { - return - } - - q.lock.Lock() - q.items = append(q.items, items...) - q.lock.Unlock() -} - -// Pop returns the head of items. -func (q *Queue) Pop() interface{} { - if len(q.items) == 0 { - return nil - } - - q.lock.Lock() - head := q.items[0] - q.items = q.items[1:] - q.lock.Unlock() - return head -} - -// First returns the head of items without deleting. -func (q *Queue) First() interface{} { - if len(q.items) == 0 { - return nil - } - - q.lock.RLock() - head := q.items[0] - q.lock.RUnlock() - return head -} - -// Copy get the copy of queue. -func (q *Queue) Copy() []interface{} { - items := make([]interface{}, 0) - q.lock.RLock() - items = append(items, q.items...) - q.lock.RUnlock() - return items -} - -// Len returns the number of items in this queue. -func (q *Queue) Len() int64 { - q.lock.Lock() - defer q.lock.Unlock() - - return int64(len(q.items)) -} - -// New is a constructor for a new concurrent safe queue. -func New(hint int64) *Queue { - return &Queue{ - items: make([]interface{}, 0, hint), - } -} diff --git a/component/session/server.go b/component/session/server.go index b3af44a..49d40db 100644 --- a/component/session/server.go +++ b/component/session/server.go @@ -13,7 +13,6 @@ import ( "sync/atomic" "time" - "github.com/xjasonlyu/tun2socks/common/queue" C "github.com/xjasonlyu/tun2socks/constant" ) @@ -25,47 +24,38 @@ var ( ) type Server struct { + sync.Mutex *http.Server trafficUp int64 trafficDown int64 - activeSessionMap sync.Map - completedSessionQueue *queue.Queue + activeSessionMap sync.Map + completedSessions []Session } func NewServer() *Server { - return &Server{ - completedSessionQueue: queue.New(maxCompletedSessions), - } + return &Server{} } func (s *Server) handler(resp http.ResponseWriter, req *http.Request) { // Slice of active sessions - var activeSessions []*Session + var activeSessions []Session s.activeSessionMap.Range(func(key, value interface{}) bool { - activeSessions = append(activeSessions, value.(*Session)) + session := value.(*Session) + activeSessions = append(activeSessions, *session) return true }) - // Slice of completed sessions - var completedSessions []*Session - for _, item := range s.completedSessionQueue.Copy() { - if session, ok := item.(*Session); ok { - completedSessions = append(completedSessions, session) - } - } + completedSessions := append([]Session(nil), s.completedSessions...) - tablePrint := func(w io.Writer, sessions []*Session) { + tablePrint := func(w io.Writer, sessions []Session) { // Sort by session start time. - sort.Slice(sessions, func(i, j int) bool { - return sessions[i].SessionStart.Sub(sessions[j].SessionStart) < 0 - }) - _, _ = fmt.Fprintf(w, "") - _, _ = fmt.Fprintf(w, "\n") sort.Slice(sessions, func(i, j int) bool { return sessions[i].SessionStart.After(sessions[j].SessionStart) }) + _, _ = fmt.Fprintf(w, "
ProcessNetworkDateDurationClient AddrTarget AddrUploadDownload
") + _, _ = fmt.Fprintf(w, "\n") for _, session := range sessions { _, _ = fmt.Fprintf(w, "\n", @@ -76,8 +66,8 @@ func (s *Server) handler(resp http.ResponseWriter, req *http.Request) { // session.DialerAddr, session.ClientAddr, session.TargetAddr, - byteCountSI(atomic.LoadInt64(&session.UploadBytes)), - byteCountSI(atomic.LoadInt64(&session.DownloadBytes)), + byteCountSI(session.UploadBytes), + byteCountSI(session.DownloadBytes), ) } _, _ = fmt.Fprintf(w, "
ProcessNetworkDateDurationClient AddrTarget AddrUploadDownload
%v%v%v%v%v%v%v%v
") @@ -103,8 +93,8 @@ table, th, td { trafficUp := atomic.LoadInt64(&s.trafficUp) trafficDown := atomic.LoadInt64(&s.trafficDown) for _, session := range activeSessions { - trafficUp += atomic.LoadInt64(&session.UploadBytes) - trafficDown += atomic.LoadInt64(&session.DownloadBytes) + trafficUp += session.UploadBytes + trafficDown += session.DownloadBytes } _, _ = fmt.Fprintf(w, "%v%v%v%v%v%v%v%v\n", date(time.Now()), @@ -176,9 +166,11 @@ func (s *Server) RemoveSession(key interface{}) { atomic.AddInt64(&s.trafficUp, atomic.LoadInt64(&session.UploadBytes)) atomic.AddInt64(&s.trafficDown, atomic.LoadInt64(&session.DownloadBytes)) // move to completed sessions - s.completedSessionQueue.Put(session) - if s.completedSessionQueue.Len() > maxCompletedSessions { - s.completedSessionQueue.Pop() + s.Lock() + s.completedSessions = append(s.completedSessions, *session) + if len(s.completedSessions) > maxCompletedSessions { + s.completedSessions = s.completedSessions[1:] } + s.Unlock() } }