Golang app-simulator-processor-components-channel.Channel类(方法)实例源码

func (this *Executor) Run(input map[info.CategoryEnum]channel.Channel, commonDataBus channel.Channel) {
	// Launch each unit as a goroutine
	unit, event := this.getUnitFromCategory(this.Category())
	logger.Print(" => Initializing execution unit (%s) %d", this.Category(), this.Index())
	go func() {
		for {
			value, running := <-input[this.Category()].Channel()
			if !running || !this.IsActive() {
				logger.Print(" => Flushing execution unit (%s) %d", this.Category(), this.Index())
				return
			}
			op := operation.Cast(value)
			for i := uint8(0); i < op.Instruction().Info.Cycles; i++ {
				this.Processor().Wait(1)
			}

			// Iterate instructions received via the channel
			op, _ = this.executeOperation(unit, event, op)
			// Send data to common bus for reservation station feedback
			commonDataBus.Add(op)
			// Release one item from input Channel
			input[this.Category()].Release()
		}
	}()
}

func (this *Dispatcher) runDispatcherToReservationStation(input channel.Channel,
	rs *reservationstation.ReservationStation, rat *registeraliastable.RegisterAliasTable, rob *reorderbuffer.ReorderBuffer) {

	incomingQueue := map[uint32]*operation.Operation{}
	currentOperationId := this.StartOperationId()

	// For each operation received to schedule, process it
	for {
		value, running := <-input.Channel()
		if !running || !this.IsActive() {
			logger.Print(" => Flushing dispatcher unit %d (dispatcher to RS)", this.Index())
			return
		}
		op := operation.Cast(value)

		// Add to current operation
		incomingQueue[op.Id()] = op

		// Send to incoming channel pending ops (if available)
		for op, exists := incomingQueue[currentOperationId]; exists; op, exists = incomingQueue[currentOperationId] {

			// Allocate in ROB if there is spacde, otherwise stall
			rob.Allocate(op)

			// Rename register in case of WAR & WAR hazards
			if this.RegisterAliasTableEntries() > 0 {
				_, destRegister := rs.GetDestinationDependency(op.Id(), op.Instruction())
				if destRegister != -1 {
					found, _ := rat.AddMap(uint32(destRegister), op.Id())
					if !found {
						// Need to stall for an available RAT entry
						logger.Collect(" => [DI%d][%03d]: No entry available in RAT. Wait for one...", this.Index(), op.Id())
						break
					}

					// Rename to physical registers
					this.renameRegisters(op.Id(), op, rat)
				}
			}

			//Redirect input operations to the required execution unit channels
			logger.Collect(" => [DI%d][%03d]: Scheduling to RS: %s, %s", this.Index(), op.Id(), op.Instruction().Info.ToString(), op.Instruction().Data.ToString())
			rs.Schedule(op)
			currentOperationId += 1
		}
		input.Release()
	}
}

func (this *Processor) RunRecovery(recoveryChannel channel.Channel, flushFunc func()) {
	for value := range recoveryChannel.Channel() {
		op := operation.Cast(value)
		logger.Collect(" => Recovering at OpId: %d and Address: %#04X", op.Id(), op.Address())

		logger.SetVerboseQuiet(true)
		// Flush pipeline
		flushFunc()
		// Clean logs
		this.RemoveForwardLogs(op.Id() - 1)
		// Clear speculative jumps
		this.ClearSpeculativeJumps()
		// Start pipeline from the recovery address
		flushFunc = this.StartPipelineUnits(this.Config(), recoveryChannel, op.Id(), op.Address())
		// Release value from channel
		recoveryChannel.Release()
	}
}

func (this *ReservationStation) runCommonBusListener(commonDataBus channel.Channel) {
	// For each operation executed, feed reservation station to release operands
	for {
		value, running := <-commonDataBus.Channel()
		if !running {
			this.reservationStation.isActive = false
			logger.Print(" => Flushing reservation station unit %d (CDB listener)", this.Index())
			return
		}

		op := operation.Cast(value)
		commonDataBus.Release()
		this.Lock() <- true

		dest, _, _ := this.getComponentsFromInstruction(op.Instruction())
		entryIndex := this.getEntryIndexFromOperationId(op.Id())
		entryOp := this.Entries()[entryIndex]
		logger.Collect(" => [RS%d][%03d]: Operation completed, releasing entry %d", this.Index(), op.Id(), entryIndex)
		if entryIndex != INVALID_INDEX {
			// Release entry
			this.Entries()[entryIndex].Busy = false
			this.Entries()[entryIndex].Free = true
			// Release entry from reservation station queue
			this.Input().Release()
		}
		// Release destination register (as RAT if enabled)
		if dest != INVALID_INDEX {
			logger.Collect(" => [RS%d][%03d]: Register %v resolved", this.Index(), op.Id(), entryOp.Destination)
			this.releaseOperation(entryOp.Destination)
		}

		// Release operands registers
		for _, operand := range entryOp.Operands {
			if operand.IsValid() && (operand.Type == MemoryType || len(this.RegisterAliasTable().Entries()) == 0) {
				logger.Collect(" => [RS%d][%03d]: Register %v resolved", this.Index(), op.Id(), operand)
				this.releaseOperation(operand)
			}
		}
		<-this.Lock()
	}
}

func (this *Decoder) Run(input, output channel.Channel) {
	// Launch each unit as a goroutine
	logger.Print(" => Initializing decoder unit %d", this.Index())
	go func() {
		for {
			value, running := <-input.Channel()
			if !running || !this.IsActive() {
				logger.Print(" => Flushing decoder unit %d", this.Index())
				return
			}
			op := operation.Cast(value)
			// Iterate instructions received via the channel
			instruction, err := this.decodeInstruction(op)
			if err != nil {
				logger.Error(err.Error())
				break
			}
			// Send data to output
			op.SetInstruction(instruction)
			output.Add(op)
			// Release one item from input Channel
			input.Release()
		}
	}()
}

func (this *Dispatcher) Run(input channel.Channel, output map[info.CategoryEnum]channel.Channel, commonDataBus, recoveryBus channel.Channel) {

	// Create register alias table
	rat := registeraliastable.New(this.Index(), this.RegisterAliasTableEntries())

	// Create re-order buffer
	rob := reorderbuffer.New(this.Index(),
		this.Processor(),
		this.StartOperationId(),
		this.ReorderBufferEntries(),
		this.InstructionsWrittenPerCycle(),
		rat)
	commonDataBusROB := channel.New(commonDataBus.Capacity())

	// Create reservation station
	rs := reservationstation.New(this.Index(), this.Processor(),
		this.Registers(),
		this.ReservationStationEntries(),
		this.InstructionsFetchedPerCycle(),
		rat, rob.Bus())
	commonDataBusRS := channel.New(commonDataBus.Capacity())

	// Create storage bus
	this.dispatcher.bus = rob.Bus()

	// Launch each unit as a goroutine
	logger.Print(" => Initializing dispatcher unit %d", this.Index())

	// Start dispatcher of operations to be executed into reservation station
	go this.runDispatcherToReservationStation(input, rs, rat, rob)
	// Start common bus multiplexer to send ack to reservation station and reorder buffer
	go this.runCommonBusMultiplexer(commonDataBus, commonDataBusRS, commonDataBusROB)

	// Run reservation station
	rs.Run(commonDataBusRS, output)
	// Run re-order buffer
	rob.Run(commonDataBusROB, recoveryBus)
}

func (this *Fetcher) Run(input, output channel.Channel) {
	logger.Print(" => Initializing fetcher unit %d", this.Index())
	// Launch each unit as a goroutine
	go func() {
		for {
			value, running := <-input.Channel()
			if !running || !this.IsActive() {
				logger.Print(" => Flushing fetcher unit %d", this.Index())
				return
			}
			// Release item from input Channel
			input.Release()

			// Initial operation (address)
			op := operation.Cast(value)

			// Load instructions data from memory
			data := this.Processor().InstructionsMemory().Load(op.Address(), consts.BYTES_PER_WORD*this.InstructionsFetchedPerCycle())

			// Fetch instructions
			startCycles := this.Processor().Cycles()
			operations, err := this.fetchInstructions(op, data, input)
			if err != nil {
				logger.Error(err.Error())
				break
			}

			// Wait cycles of a fetch stage
			this.Processor().Wait(consts.FETCH_CYCLES)

			// After wait cycle, notify decode channel with new operations
			for _, op := range operations {
				if this.IsActive() {
					this.Processor().LogEvent(consts.FETCH_EVENT, this.Index(), op.Id(), startCycles)
					output.Add(op)
				}
			}
		}
	}()
}

func (this *Fetcher) fetchInstructions(op *operation.Operation, bytes []byte, input channel.Channel) ([]*operation.Operation, error) {

	initialAddress := op.Address()
	totalInstructions := len(bytes) / consts.BYTES_PER_WORD
	ops := []*operation.Operation{}

	// Analyze each instruction loaded
	for i := 0; i < totalInstructions; i += 1 {

		data := bytes[i*consts.BYTES_PER_WORD : (i+1)*consts.BYTES_PER_WORD]

		// Check program reach end
		if this.Processor().ReachedEnd(data) {
			this.processor.Finish()
			return ops, nil
		}

		// Do fetch once a new address is received
		msg := fmt.Sprintf(" => [FE%d][%03d]: INS[%#04X] = %#04X", this.Index(), this.Processor().InstructionsFetchedCounter(), op.Address(), data)
		value, ok := this.Processor().InstructionsMap()[op.Address()]
		if ok {
			msg = fmt.Sprintf("%s // %s", msg, strings.TrimSpace(strings.Split(value, "=>")[1]))
		}
		logger.Collect(msg)

		// Log event
		this.Processor().LogInstructionFetched(op.Address())

		// Update data into operation and add to array for post-events
		op.SetWord([]byte{data[0], data[1], data[2], data[3]})

		// Add operation to be sent to decode channel
		ops = append(ops, op)

		// Do pre-decode
		needsWait, instruction := this.BranchPredictor().PreDecodeInstruction(op.Address())

		// If is not pipelined than wait instruction to finish
		if !this.Processor().Config().Pipelined() {
			go func() {
				address, _, err := this.BranchPredictor().GetNextAddress(op.Address(), instruction, true)
				newOp := operation.New(this.Processor().InstructionsFetchedCounter(), address)
				if err == nil {
					input.Add(newOp)
				}
			}()
			return ops, nil
		}

		// Add next instruction for fetching (as many instructions as it supports per cycle)
		if needsWait {
			logger.Collect(" => [FE%d][%03d]: Wait detected, no fetching more instructions this cycle", this.Index(), this.Processor().InstructionsFetchedCounter()-1)
			// Add next instruction in a go routine as it need to be stalled
			go func() {
				address, _, err := this.BranchPredictor().GetNextAddress(op.Address(), instruction, false)
				newOp := operation.New(this.Processor().InstructionsFetchedCounter(), address)
				if err == nil {
					input.Add(newOp)
				}
			}()
			return ops, nil
		} else {
			address, predicted, err := this.BranchPredictor().GetNextAddress(op.Address(), instruction, false)
			// Set current operation added to be decoded the predicted address
			if predicted {
				ops[len(ops)-1].SetNextPredictedAddress(address)
			}

			// Create new operation object
			op = operation.New(this.Processor().InstructionsFetchedCounter(), address)
			// If is the last instruction from the package or the predicted address is outside of the address package
			if err == nil && (i >= totalInstructions-1 || initialAddress+(uint32(i+1)*consts.BYTES_PER_WORD) != op.Address()) {
				input.Add(op)
				return ops, nil
			}
		}
	}
	return ops, nil
}

func (this *Dispatcher) runCommonBusMultiplexer(input, output1, output2 channel.Channel) {
	// For each result got from execution units in the common data bus send to RS and ROB
	for {
		value, running := <-input.Channel()
		if !running {
			output1.Close()
			output2.Close()
			logger.Print(" => Flushing dispatcher unit %d (CDB Mux)", this.Index())
			return
		}
		output1.Add(value)
		output2.Add(value)
		input.Release()
	}
}

func (this *ReorderBuffer) Run(commonDataBus channel.Channel, recoveryBus channel.Channel) {
	// Launch unit as a goroutine
	logger.Print(" => Initializing re-order buffer unit %d", this.Index())
	opId := this.StartOperationId()
	misprediction := false
	clockAllowed := this.Processor().Cycles()
	forceClose := false

	go func() {
		for {
			_, running := <-commonDataBus.Channel()
			if !running || misprediction {
				forceClose = true
				logger.Print(" => Flushing re-order buffer unit %d", this.Index())
				return
			}
			commonDataBus.Release()
		}
	}()

	go func() {
		for {
			if forceClose {
				return
			}

			if this.Processor().Cycles() < clockAllowed {
				this.Processor().Wait(1)
				continue
			}

			// Commit in order, if missing an operation, wait for it
			computedAddress := uint32(0)
			robEntries := []RobEntry{}
			for robEntry, exists := this.Buffer()[opId]; exists; robEntry, exists = this.Buffer()[opId] {
				if uint32(len(robEntries)) >= this.InstructionsWrittenPerCycle() {
					break
				}
				// Ensure we can write results the next cycle result was written into ROB
				if this.Processor().Cycles() > robEntry.Cycle+1 {
					// Check for misprediction
					misprediction, computedAddress = this.checkForMisprediction(this.Buffer()[opId], robEntries)
					// Decrement speculative jumps
					this.Processor().DecrementSpeculativeJump()
					// Add to queue for commit
					robEntries = append(robEntries, robEntry)
					opId += 1
					// If misprediction, do not process more rob entries
					if misprediction {
						break
					}
				}
			}
			this.commitRobEntries(robEntries)
			if misprediction {
				this.Processor().Wait(consts.WRITEBACK_CYCLES)
				recoveryBus.Add(operation.New(opId, computedAddress))
			}
			clockAllowed = this.Processor().Cycles() + 1
		}
	}()
}