探索协程世界：原理、现实与漏洞模式（下）

def get_async_backend(asynclib_name: str | None = None) -> AsyncBackend:
    if asynclib_name is None:
        asynclib_name = sniffio.current_async_library()

    modulename = "anyio._backends._" + asynclib_name
    try:
        module = sys.modules[modulename]
    except KeyError:
        module = import_module(modulename)

    return getattr(module, "backend_class")

async def sleep(delay: float) -> None:
    """
    Pause the current task for the specified duration.
    :param delay: the duration, in seconds
    """
    return await get_async_backend().sleep(delay)

async def func(d):

    await asy().bfunc()

    try:
        module = sys.modules[modulename]
    except KeyError:
        module = import_module(modulename)

try:
    module = sys.modules[modulename] ## <= 线程2，刚开始访问这就modules
except KeyError:
    module = import_module(modulename) ## <= 线程1，在这个函数内部

spec._initializing = True
try:
    sys.modules[spec.name] = module ## <= 线程1，假设刚经过这里
    try:
        if spec.loader is None:
            if spec.submodule_search_locations is None:
                raise ImportError('missing loader', name=spec.name)
            ## A namespace package so do nothing.
        else:
            spec.loader.exec_module(module) ## 线程1还未来得及执行module的初始化函数
    except:

- def get_async_backend(asynclib_name: str | None = None) -> AsyncBackend:
+ def get_async_backend(asynclib_name: str | None = None) -> type[AsyncBackend]:
    if asynclib_name is None:
        asynclib_name = sniffio.current_async_library()

-    modulename = "anyio._backends._" + asynclib_name
+    ## We use our own dict instead of sys.modules to get the already imported back-end
+    ## class because the appropriate modules in sys.modules could potentially be only
+    ## partially initialized
    try:
-        module = sys.modules[modulename]
+        return loaded_backends[asynclib_name]
    except KeyError:
-        module = import_module(modulename)
-
-    return getattr(module, "backend_class")
+        module = import_module(f"anyio._backends._{asynclib_name}")
+        loaded_backends[asynclib_name] = module.backend_class
+        return module.backend_class

await get_async_backend().sleep(delay)

use dashmap::DashMap;
use futures::future::join_all;
use std::time::Duration;
use tokio;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let tasks: DashMap<i32, i32> = DashMap::new();
    let t = &tasks;
    let mut async_events = vec![];
    for i in 0..30 {
        tasks.insert(i, i);
        async_events.push(async move {
            let j = i;
            if let Some(_) = t.get(&j) {
                tokio::time::delay_for(Duration::from_secs(1)).await;
            }

            if let Some(_) = t.get_mut(&j) {
                tokio::time::delay_for(Duration::from_secs(1)).await;
            }
        });
    }
    join_all(async_events).await;
    println!("Hello, world!");
    Ok(())
}

if let Some(_) = t.get_mut(&j)
{
    match #[lang = "into_future"](tokio::time::sleep(Duration::from_secs(1)))
        {
        mut __awaitee =>
        loop {
            match unsafe {
                    #[lang = "poll"](#[lang = "new_unchecked"](&mut __awaitee),
                        #[lang = "get_context"](_task_context))
                    } {
                #[lang = "Ready"] {  0: result } => break result,
                #[lang = "Pending"] {} => { }    
                }
            _task_context = (yield ());
            },
    };
}

let mut t_value = 0;
// 先将数据拿出来
if let Some(value) = t.remove(&j) {
    t_value = value.1;
}
// 在外部处理完成
tokio::time::sleep(Duration::from_secs(1)).await;
// 最后在塞回去
t.insert(j, t_value);

package main
import (
	"fmt"
	"sync"
)

func main() {
	ch := make(chan int) // 无缓冲通道
	var wg sync.WaitGroup

	// 启动多个 goroutine
	for i := 0; i < 3; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			fmt.Printf("Goroutine %d: Sending data to channel...\n", id)
			ch <- id // 阻塞在这里
			fmt.Printf("Goroutine %d: Finished sending data.\n", id)
		}(i)
	}

	fmt.Println("Main: Waiting for goroutines to finish...")
	wg.Wait()
}

func createStreams(req *http.Request, w http.ResponseWriter, opts *Options, supportedStreamProtocols []string, idleTimeout, streamCreationTimeout time.Duration) (*context, bool) {
	var ctx *context
	var ok bool
	if wsstream.IsWebSocketRequest(req) {
		ctx, ok = createWebSocketStreams(req, w, opts, idleTimeout)
	} else {
		ctx, ok = createHTTPStreamStreams(req, w, opts, supportedStreamProtocols, idleTimeout, streamCreationTimeout)
	}
	if !ok {
		return nil, false
	}
	if ctx.resizeStream != nil {
		ctx.resizeChan = make(chan remotecommand.TerminalSize)
		go handleResizeEvents(ctx.resizeStream, ctx.resizeChan)
	}

	return ctx, true
}

func handleResizeEvents(stream io.Reader, channel chan<- remotecommand.TerminalSize) {
	defer runtime.HandleCrash()
	defer close(channel)

	decoder := json.NewDecoder(stream)
	for {
		size := remotecommand.TerminalSize{}
		if err := decoder.Decode(&size); err != nil {
			break
		}
		channel <- size     // 注意这里！
	}
}

// handleResizing spawns a goroutine that processes the resize channel, calling resizeFunc for each
// remotecommand.TerminalSize received from the channel.
func handleResizing(resize <-chan remotecommand.TerminalSize, resizeFunc func(size remotecommand.TerminalSize)) {
	if resize == nil {
		return
	}

	go func() {
		defer runtime.HandleCrash()

		for {
			size, ok := <-resize
			if !ok {
                return
            }

			if size.Height < 1 || size.Width < 1 {
				continue
			}
			resizeFunc(size)
		}
	}()
}

// execInContainer executes a command inside the container synchronously, and
// redirects stdio stream properly.
// This function only returns when the exec process exits, this means that:
// ---------------------- 留意这里 ------------------------------------
// 1) As long as the exec process is running, the goroutine in the cri plugin
// will be running and wait for the exit code;
// --------------------------------------------------------------------
func (c *criService) execInContainer(ctx context.Context, id string, opts execOptions) (*uint32, error) {
	span := tracing.SpanFromContext(ctx)
	// Get container from our container store.
	cntr, err := c.containerStore.Get(id)

	if err != nil {
		return nil, fmt.Errorf("failed to find container %q in store: %w", id, err)
	}
	id = cntr.ID
	span.SetAttributes(tracing.Attribute("container.id", id))

	state := cntr.Status.Get().State()
	if state != runtime.ContainerState_CONTAINER_RUNNING {
		return nil, fmt.Errorf("container is in %s state", criContainerStateToString(state))
	}

	return c.execInternal(ctx, cntr.Container, id, opts)
}


func (c *criService) execInternal(ctx context.Context, container containerd.Container, id string, opts execOptions) (*uint32, error) {
	// 省略一些别的代码
	spec, err := container.Spec(ctx)
	if err != nil {
		return nil, fmt.Errorf("failed to get container spec: %w", err)
	}
	task, err := container.Task(ctx, nil)
	if err != nil {
		return nil, fmt.Errorf("failed to load task: %w", err)
	}
	pspec := spec.Process

	pspec.Terminal = opts.tty
	if opts.tty {
		if err := oci.WithEnv([]string{"TERM=xterm"})(ctx, nil, nil, spec); err != nil {
			return nil, fmt.Errorf("add TERM env var to spec: %w", err)
		}
	}

	pspec.Args = opts.cmd
	// CommandLine may already be set on the container's spec, but we want to only use Args here.
	pspec.CommandLine = ""

	if opts.stdout == nil {
		opts.stdout = cio.NewDiscardLogger()
	}
	if opts.stderr == nil {
		opts.stderr = cio.NewDiscardLogger()
	}
	execID := util.GenerateID()
	log.G(ctx).Debugf("Generated exec id %q for container %q", execID, id)
	volatileRootDir := c.getVolatileContainerRootDir(id)
	var execIO *cio.ExecIO

    // 这里执行指令
	process, err := task.Exec(ctx, execID, pspec,
		func(id string) (containerdio.IO, error) {
			cntr, err := c.containerStore.Get(container.ID())
			if err != nil {
				return nil, fmt.Errorf("an error occurred when try to find container %q: %w", container.ID(), err)
			}
			sb, err := c.sandboxStore.Get(cntr.SandboxID)
			if err != nil {
				return nil, fmt.Errorf("an error occurred when try to find sandbox %q: %w", cntr.SandboxID, err)
			}
			ociRuntime, err := c.config.GetSandboxRuntime(sb.Config, sb.Metadata.RuntimeHandler)
			if err != nil {
				return nil, fmt.Errorf("failed to get sandbox runtime: %w", err)
			}
			switch ociRuntime.IOType {
			case config.IOTypeStreaming:
				execIO, err = cio.NewStreamExecIO(id, sb.Endpoint.Address, opts.tty, opts.stdin != nil)
			default:
				execIO, err = cio.NewFifoExecIO(id, volatileRootDir, opts.tty, opts.stdin != nil)
			}

			return execIO, err
		},
	)
	
    // 调用resize函数
	handleResizing(ctx, opts.resize, func(size remotecommand.TerminalSize) {
		if err := process.Resize(ctx, uint32(size.Width), uint32(size.Height)); err != nil {
			log.G(ctx).WithError(err).Errorf("Failed to resize process %q console for container %q", execID, id)
		}
	})

}

name: envoy.filters.http.lua
typed_config:
  "@type": type.googleapis.com/envoy.extensions.filters.http.lua.v3.Lua
  default_source_code:
    inline_string: |
      -- Called on the request path.
      function envoy_on_request(request_handle)
        -- Do something.
      end
      -- Called on the response path.
      function envoy_on_response(response_handle)
        -- Do something.
      end

// envoy/source/common/http/filter_manager.cc
// 这里负责主动设置接受数据的上下限，并且给出当数据超出上线/下线的时候，会调用的对应处理函数
// 在这里会限制发送数据和返回数据的大小限制
void FilterManager::setBufferLimit(uint32_t new_limit) {
  ENVOY_STREAM_LOG(debug, "setting buffer limit to {}", *this, new_limit);
  buffer_limit_ = new_limit;
  if (buffered_request_data_) {
    buffered_request_data_->setWatermarks(buffer_limit_);
  }
  if (buffered_response_data_) {
    buffered_response_data_->setWatermarks(buffer_limit_);
  }
}
// 这里是创建对应给Filter使用的buffer对象数据的时候，设置传输数据的边界
Buffer::InstancePtr ActiveStreamDecoderFilter::createBuffer() {
  auto buffer = dispatcher().getWatermarkFactory().createBuffer(
      [this]() -> void { this->requestDataDrained(); },
      [this]() -> void { this->requestDataTooLarge(); }, // above_high_watermark_
      []() -> void { /* TODO(adisuissa): Handle overflow watermark */ });
  buffer->setWatermarks(parent_.buffer_limit_);
  return buffer;
}


// 当接收到的数据太大的时候，会直接发送返回数据
void ActiveStreamDecoderFilter::requestDataTooLarge() {
  ENVOY_STREAM_LOG(debug, "request data too large watermark exceeded", parent_);
  if (parent_.state_.decoder_filters_streaming_) {
    onDecoderFilterAboveWriteBufferHighWatermark();
  } else {
    parent_.filter_manager_callbacks_.onRequestDataTooLarge();
    sendLocalReply(Code::PayloadTooLarge, CodeUtility::toString(Code::PayloadTooLarge), nullptr,
                   absl::nullopt, StreamInfo::ResponseCodeDetails::get().RequestPayloadTooLarge);
  }
}
//发送返回数据
void ActiveStreamFilterBase::sendLocalReply(
    Code code, absl::string_view body,
    std::function<void(ResponseHeaderMap& headers)> modify_headers,
    const absl::optional<Grpc::Status::GrpcStatus> grpc_status, absl::string_view details) {
  if (!streamInfo().filterState()->hasData<LocalReplyOwnerObject>(LocalReplyFilterStateKey)) {
    streamInfo().filterState()->setData(
        LocalReplyFilterStateKey,
        std::make_shared<LocalReplyOwnerObject>(filter_context_.config_name),
        StreamInfo::FilterState::StateType::ReadOnly,
        StreamInfo::FilterState::LifeSpan::FilterChain);
  }

  parent_.sendLocalReply(code, body, modify_headers, grpc_status, details);
}

// envoy/source/common/buffer/watermark_buffer.cc
// 这边会设置接受数据的上限和下限，然后调用对应的处理函数
// 这里负责记录数据的上限
 // 默认定义函数，这里的setWatermarks第二个参数可以是0
 void setWatermarks(uint32_t high_watermark, uint32_t overflow_watermark = 0) override;

void WatermarkBuffer::setWatermarks(uint32_t high_watermark,
                                    uint32_t overflow_watermark_multiplier) {
  if (overflow_watermark_multiplier > 0 &&
      (static_cast<uint64_t>(overflow_watermark_multiplier) * high_watermark) >
          std::numeric_limits<uint32_t>::max()) {
    ENVOY_LOG_MISC(debug, "Error setting overflow threshold: overflow_watermark_multiplier * "
                          "high_watermark is overflowing. Disabling overflow watermark.");
    overflow_watermark_multiplier = 0;
  }
  low_watermark_ = high_watermark / 2;
  high_watermark_ = high_watermark;
  overflow_watermark_ = overflow_watermark_multiplier * high_watermark;
  checkHighAndOverflowWatermarks();
  checkLowWatermark();
}

// 这里会检查接受的数据是否过多/过少
void WatermarkBuffer::checkHighAndOverflowWatermarks() {
  if (high_watermark_ == 0 || OwnedImpl::length() <= high_watermark_) {
    return;
  }

  if (!above_high_watermark_called_) {
    above_high_watermark_called_ = true;
    above_high_watermark_();
  }

  // Check if overflow watermark is enabled, wasn't previously triggered,
  // and the buffer size is above the threshold
  if (overflow_watermark_ != 0 && !above_overflow_watermark_called_ &&
      OwnedImpl::length() > overflow_watermark_) {
    above_overflow_watermark_called_ = true;
    above_overflow_watermark_();
  }
}

Http::FilterDataStatus StreamHandleWrapper::onData(Buffer::Instance& data, bool end_stream) {

  ASSERT(!end_stream_);
  end_stream_ = end_stream;
  saw_body_ = true;

  if (state_ == State::WaitForBodyChunk) {
    ENVOY_LOG(trace, "resuming for next body chunk");
    Filters::Common::Lua::LuaDeathRef<Filters::Common::Lua::BufferWrapper> wrapper(
        Filters::Common::Lua::BufferWrapper::create(coroutine_.luaState(), headers_, data), true);
    state_ = State::Running;
    coroutine_.resume(1, yield_callback_);
  } else if (state_ == State::WaitForBody && end_stream_) {
    ENVOY_LOG(debug, "resuming body due to end stream");
    callbacks_.addData(data); // 注意这里
    state_ = State::Running;
    coroutine_.resume(luaBody(coroutine_.luaState()), yield_callback_);
  } else if (state_ == State::WaitForTrailers && end_stream_) {
    ENVOY_LOG(debug, "resuming nil trailers due to end stream");
    state_ = State::Running;
    coroutine_.resume(0, yield_callback_);
  }
 if (state_ == State::HttpCall || state_ == State::WaitForBody) {
    ENVOY_LOG(trace, "buffering body");
    return Http::FilterDataStatus::StopIterationAndBuffer;
  } else if (state_ == State::Responded) {
    return Http::FilterDataStatus::StopIterationNoBuffer;
  } else {
    headers_continued_ = true;
    return Http::FilterDataStatus::Continue;
  }
}

void PlatformBridgeFilter::RequestFilterBase::addData(envoy_data data) {
  Buffer::OwnedImpl inject_data;
  inject_data.addBufferFragment(*Buffer::BridgeFragment::createBridgeFragment(data));
  parent_.decoder_callbacks_->addDecodedData(inject_data, /* watermark */ false);
}

void FilterManager::addDecodedData(ActiveStreamDecoderFilter& filter, Buffer::Instance& data,
                                   bool streaming) {
  if (state_.filter_call_state_ == 0 ||
      (state_.filter_call_state_ & FilterCallState::DecodeHeaders) ||
      (state_.filter_call_state_ & FilterCallState::DecodeData) ||
      ((state_.filter_call_state_ & FilterCallState::DecodeTrailers) && !filter.canIterate())) {
    // Make sure if this triggers watermarks, the correct action is taken.
    state_.decoder_filters_streaming_ = streaming;
    // If no call is happening or we are in the decode headers/data callback, buffer the data.
    // Inline processing happens in the decodeHeaders() callback if necessary.
    filter.commonHandleBufferData(data);
    // 省略部分代码
      }
}

void ActiveStreamFilterBase::commonHandleBufferData(Buffer::Instance& provided_data) {

  // The way we do buffering is a little complicated which is why we have this common function
  // which is used for both encoding and decoding. When data first comes into our filter pipeline,
  // we send it through. Any filter can choose to stop iteration and buffer or not. If we then
  // continue iteration in the future, we use the buffered data. A future filter can stop and
  // buffer again. In this case, since we are already operating on buffered data, we don't
  // rebuffer, because we assume the filter has modified the buffer as it wishes in place.
  if (bufferedData().get() != &provided_data) {
    if (!bufferedData()) {
      bufferedData() = createBuffer();  //可以看到这里创建了对应的buffer对象
    }
    bufferedData()->move(provided_data);
  }
}

void Filter::onDestroy() {
  destroyed_ = true;
  if (request_stream_wrapper_.get()) {
    request_stream_wrapper_.get()->onReset();
  }
  if (response_stream_wrapper_.get()) {
    response_stream_wrapper_.get()->onReset();
  }
}

R"EOF(
name: lua
typed_config:
  "@type": type.googleapis.com/envoy.extensions.filters.http.lua.v3.Lua
  default_source_code:
    inline_string: |
      function envoy_on_request(request_handle)
        local initial_req_body = request_handle:body()
        local headers, body = request_handle:httpCall(
        "lua_cluster",
        {
          [":method"] = "POST",
          [":path"] = "/",
          [":authority"] = "lua_cluster"
        },
        "hello world",
        1000)
        request_handle:headers():replace("x-code", headers["code"] or "")
      end
)EOF";

local initial_req_body = request_handle:body()

const int buffer_limit = 65535;
config_helper_.setBufferLimits(buffer_limit, buffer_limit);

initializeFilter(FILTER_AND_CODE);
codec_client_ = makeHttpConnection(lookupPort("http"));
auto encoder_decoder =
    codec_client_->startRequest(Http::TestRequestHeaderMapImpl{{":method", "POST"},
                                                                {":scheme", "http"},
                                                                {":path", "/test/long/url"},
                                                                {":authority", "host"}});
auto request_encoder = &encoder_decoder.first;
auto response = std::move(encoder_decoder.second);

codec_client_->sendData(*request_encoder, buffer_limit + 1, true);

void onReset() {
    if (http_request_) {
        http_request_->cancel();
        http_request_ = nullptr;
    }
}

Http::FilterDataStatus StreamHandleWrapper::onData(Buffer::Instance& data, bool end_stream) {

    callbacks_.addData(data); // 来到前面提到的添加数据的部分逻辑，因为数据过大，此时数据发生错误，但是只是被标记为可消除
    state_ = State::Running;
    coroutine_.resume(luaBody(coroutine_.luaState()), yield_callback_); // 此时调用协程，回到正常Lua脚本处理逻辑中
}

local headers, body = request_handle:httpCall(
        "lua_cluster",
        {
          [":method"] = "POST",
          [":path"] = "/",
          [":authority"] = "lua_cluster"
        },
        "hello world",
        1000)