Subversion Repositories openrisc

// Copyright 2010 The Go Authors.  All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package textproto

import (
        "sync"
)

// A Pipeline manages a pipelined in-order request/response sequence.
//
// To use a Pipeline p to manage multiple clients on a connection,
// each client should run:
//
//      id := p.Next()  // take a number
//
//      p.StartRequest(id)      // wait for turn to send request
//      «send request»
//      p.EndRequest(id)        // notify Pipeline that request is sent
//
//      p.StartResponse(id)     // wait for turn to read response
//      «read response»
//      p.EndResponse(id)       // notify Pipeline that response is read
//
// A pipelined server can use the same calls to ensure that
// responses computed in parallel are written in the correct order.
type Pipeline struct {
        mu       sync.Mutex
        id       uint
        request  sequencer
        response sequencer
}

// Next returns the next id for a request/response pair.
func (p *Pipeline) Next() uint {
        p.mu.Lock()
        id := p.id
        p.id++
        p.mu.Unlock()
        return id
}

// StartRequest blocks until it is time to send (or, if this is a server, receive)
// the request with the given id.
func (p *Pipeline) StartRequest(id uint) {
        p.request.Start(id)
}

// EndRequest notifies p that the request with the given id has been sent
// (or, if this is a server, received).
func (p *Pipeline) EndRequest(id uint) {
        p.request.End(id)
}

// StartResponse blocks until it is time to receive (or, if this is a server, send)
// the request with the given id.
func (p *Pipeline) StartResponse(id uint) {
        p.response.Start(id)
}

// EndResponse notifies p that the response with the given id has been received
// (or, if this is a server, sent).
func (p *Pipeline) EndResponse(id uint) {
        p.response.End(id)
}

// A sequencer schedules a sequence of numbered events that must
// happen in order, one after the other.  The event numbering must start
// at 0 and increment without skipping.  The event number wraps around
// safely as long as there are not 2^32 simultaneous events pending.
type sequencer struct {
        mu   sync.Mutex
        id   uint
        wait map[uint]chan uint
}

// Start waits until it is time for the event numbered id to begin.
// That is, except for the first event, it waits until End(id-1) has
// been called.
func (s *sequencer) Start(id uint) {
        s.mu.Lock()
        if s.id == id {
                s.mu.Unlock()
                return
        }
        c := make(chan uint)
        if s.wait == nil {
                s.wait = make(map[uint]chan uint)
        }
        s.wait[id] = c
        s.mu.Unlock()
        <-c
}

// End notifies the sequencer that the event numbered id has completed,
// allowing it to schedule the event numbered id+1.  It is a run-time error
// to call End with an id that is not the number of the active event.
func (s *sequencer) End(id uint) {
        s.mu.Lock()
        if s.id != id {
                panic("out of sync")
        }
        id++
        s.id = id
        if s.wait == nil {
                s.wait = make(map[uint]chan uint)
        }
        c, ok := s.wait[id]
        if ok {
                delete(s.wait, id)
        }
        s.mu.Unlock()
        if ok {
                c <- 1
        }
}