code: def test_sync_async_mix_regression_test(shutdown_only):
      """Verify when sync and async tasks are mixed up
      it doesn't raise a segfault
  
      https://github.com/ray-project/ray/issues/41346
      """
  
      class PayloadPydantic(BaseModel):
          class Error(BaseModel):
              msg: str
              code: int
              type: str
  
          text: Optional[str] = None
          ts: Optional[float] = None
          reason: Optional[str] = None
          error: Optional[Error] = None
  
      ray.init()
  
      @ray.remote
      class B:
          def __init__(self, a):
              self.a = a
  
          async def stream(self):
              async for ref in self.a.stream.remote(1):
                  print("stream")
                  await ref
  
          async def start(self):
              await asyncio.gather(*[self.stream() for _ in range(2)])
  
      @ray.remote
      class A:
          def stream(self, i):
              payload = PayloadPydantic(
                  text="Test output",
                  ts=time.time(),
                  reason="Success!",
              )
  
              for _ in range(10):
                  yield payload
  
          async def aio_stream(self):
              for _ in range(10):
                  yield 1
  
      a = A.remote()
      b = B.remote(a)
      ray.get(b.start.remote())
  
file_path: ray/python/ray/tests/test_streaming_generator_4.py
imports: import pytest
  import numpy
  import sys
  import time
  import gc
  import random
  import asyncio
  from typing import Optional
  from pydantic import BaseModel
  import ray
label_desc: Use image node_id 4 for calling a global function with example usage: test_sync_async_mix_regression_test(shutdown_only) without return types