{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# SGE (Sun Grid Engine) Tutorial\n", "\n", "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/ContextLab/clustrix/blob/master/docs/notebooks/sge_tutorial.ipynb)\n", "\n", "This tutorial demonstrates how to use Clustrix with SGE (Sun Grid Engine) clusters, including Open Grid Scheduler and other SGE-compatible systems.\n", "\n", "## Prerequisites\n", "\n", "- Access to an SGE cluster\n", "- SSH key configured for the cluster\n", "- Clustrix installed: `pip install clustrix`" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Install Clustrix (uncomment if needed)\n", "# !pip install clustrix\n", "\n", "import clustrix\n", "from clustrix import cluster, configure\n", "import numpy as np" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## SGE Configuration\n", "\n", "Configure Clustrix for your SGE cluster:" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Configure for SGE cluster\n", "configure(\n", " cluster_type=\"sge\",\n", " cluster_host=\"sge-cluster.org\", # Replace with your cluster\n", " username=\"your-username\", # Replace with your username\n", " key_file=\"~/.ssh/id_rsa\", # SSH key path\n", " \n", " # SGE resource defaults\n", " default_cores=4,\n", " default_memory=\"8GB\",\n", " default_time=\"02:00:00\",\n", " default_queue=\"all.q\", # Common SGE queue name\n", " \n", " # SGE-specific settings\n", " remote_work_dir=\"/home/your-username/clustrix\",\n", " \n", " # Environment modules\n", " module_loads=[\"python/3.9\"],\n", " \n", " # Job management\n", " cleanup_on_success=True,\n", " max_parallel_jobs=30\n", ")\n", "\n", "print(\"SGE cluster configured successfully!\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Example 1: Mathematical Optimization\n", "\n", "SGE clusters are often used for optimization problems:" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "@cluster(\n", " cores=8, \n", " memory=\"16GB\", \n", " time=\"01:30:00\", \n", " queue=\"all.q\",\n", " pe=\"smp 8\" # SGE parallel environment\n", ")\n", "def genetic_algorithm_optimization(problem_size=1000, generations=500):\n", " \"\"\"\n", " Genetic Algorithm optimization on SGE cluster.\n", " \"\"\"\n", " import numpy as np\n", " import random\n", " from functools import partial\n", " \n", " def rastrigin_function(x):\n", " \"\"\"Rastrigin function - a multimodal optimization benchmark\"\"\"\n", " A = 10\n", " n = len(x)\n", " return A * n + sum(xi**2 - A * np.cos(2 * np.pi * xi) for xi in x)\n", " \n", " def rosenbrock_function(x):\n", " \"\"\"Rosenbrock function - another optimization benchmark\"\"\"\n", " return sum(100 * (x[i+1] - x[i]**2)**2 + (1 - x[i])**2 for i in range(len(x)-1))\n", " \n", " def sphere_function(x):\n", " \"\"\"Simple sphere function\"\"\"\n", " return sum(xi**2 for xi in x)\n", " \n", " # Choose objective function\n", " objective_functions = {\n", " 'rastrigin': rastrigin_function,\n", " 'rosenbrock': rosenbrock_function,\n", " 'sphere': sphere_function\n", " }\n", " \n", " objective_name = 'rastrigin' # Can be parameterized\n", " objective_func = objective_functions[objective_name]\n", " \n", " # Problem dimensions\n", " dimensions = min(50, problem_size // 20) # Scale dimensions with problem size\n", " bounds = (-5.12, 5.12) if objective_name == 'rastrigin' else (-2.0, 2.0)\n", " \n", " print(f\"Optimizing {objective_name} function in {dimensions} dimensions\")\n", " print(f\"Population size: {problem_size}, Generations: {generations}\")\n", " \n", " class Individual:\n", " def __init__(self, genes=None):\n", " if genes is None:\n", " self.genes = np.random.uniform(bounds[0], bounds[1], dimensions)\n", " else:\n", " self.genes = genes.copy()\n", " self.fitness = None\n", " \n", " def evaluate(self):\n", " if self.fitness is None:\n", " self.fitness = objective_func(self.genes)\n", " return self.fitness\n", " \n", " def mutate(self, mutation_rate=0.1, mutation_strength=0.1):\n", " if random.random() < mutation_rate:\n", " # Add Gaussian noise\n", " mutation = np.random.normal(0, mutation_strength, dimensions)\n", " self.genes = np.clip(self.genes + mutation, bounds[0], bounds[1])\n", " self.fitness = None # Reset fitness\n", " \n", " def crossover(self, other):\n", " # Uniform crossover\n", " mask = np.random.random(dimensions) < 0.5\n", " child1_genes = np.where(mask, self.genes, other.genes)\n", " child2_genes = np.where(mask, other.genes, self.genes)\n", " return Individual(child1_genes), Individual(child2_genes)\n", " \n", " # Initialize population\n", " population = [Individual() for _ in range(problem_size)]\n", " \n", " # Evaluate initial population\n", " for individual in population:\n", " individual.evaluate()\n", " \n", " # Evolution statistics\n", " best_fitness_history = []\n", " average_fitness_history = []\n", " diversity_history = []\n", " \n", " # Main evolution loop\n", " for generation in range(generations):\n", " if generation % (generations // 10) == 0:\n", " print(f\"Generation {generation}/{generations}\")\n", " \n", " # Selection (tournament selection)\n", " def tournament_selection(pop, tournament_size=3):\n", " tournament = random.sample(pop, tournament_size)\n", " return min(tournament, key=lambda ind: ind.evaluate())\n", " \n", " # Create new population\n", " new_population = []\n", " \n", " # Elitism - keep best 10%\n", " elite_size = max(1, problem_size // 10)\n", " elite = sorted(population, key=lambda ind: ind.evaluate())[:elite_size]\n", " new_population.extend([Individual(ind.genes) for ind in elite])\n", " \n", " # Generate offspring\n", " while len(new_population) < problem_size:\n", " parent1 = tournament_selection(population)\n", " parent2 = tournament_selection(population)\n", " \n", " if random.random() < 0.8: # Crossover probability\n", " child1, child2 = parent1.crossover(parent2)\n", " else:\n", " child1, child2 = Individual(parent1.genes), Individual(parent2.genes)\n", " \n", " # Adaptive mutation rate\n", " mutation_rate = 0.1 * (1 + generation / generations)\n", " child1.mutate(mutation_rate=mutation_rate)\n", " child2.mutate(mutation_rate=mutation_rate)\n", " \n", " new_population.extend([child1, child2])\n", " \n", " # Trim to exact population size\n", " new_population = new_population[:problem_size]\n", " population = new_population\n", " \n", " # Evaluate new population\n", " for individual in population:\n", " individual.evaluate()\n", " \n", " # Statistics\n", " fitnesses = [ind.fitness for ind in population]\n", " best_fitness = min(fitnesses)\n", " average_fitness = np.mean(fitnesses)\n", " \n", " # Population diversity (average pairwise distance)\n", " if generation % 10 == 0: # Calculate diversity every 10 generations\n", " sample_size = min(100, problem_size)\n", " sample_pop = random.sample(population, sample_size)\n", " distances = []\n", " for i in range(len(sample_pop)):\n", " for j in range(i+1, len(sample_pop)):\n", " dist = np.linalg.norm(sample_pop[i].genes - sample_pop[j].genes)\n", " distances.append(dist)\n", " diversity = np.mean(distances) if distances else 0\n", " diversity_history.append(diversity)\n", " \n", " best_fitness_history.append(best_fitness)\n", " average_fitness_history.append(average_fitness)\n", " \n", " # Final results\n", " best_individual = min(population, key=lambda ind: ind.evaluate())\n", " \n", " return {\n", " 'objective_function': objective_name,\n", " 'dimensions': dimensions,\n", " 'population_size': problem_size,\n", " 'generations': generations,\n", " 'best_fitness': best_individual.fitness,\n", " 'best_solution': best_individual.genes.tolist(),\n", " 'convergence_history': {\n", " 'best_fitness': best_fitness_history[::10], # Every 10th generation\n", " 'average_fitness': average_fitness_history[::10],\n", " 'diversity': diversity_history\n", " },\n", " 'final_population_stats': {\n", " 'best_fitness': min(fitnesses),\n", " 'worst_fitness': max(fitnesses),\n", " 'average_fitness': np.mean(fitnesses),\n", " 'fitness_std': np.std(fitnesses)\n", " }\n", " }\n", "\n", "# Run genetic algorithm optimization\n", "ga_results = genetic_algorithm_optimization(problem_size=500, generations=200)\n", "\n", "print(f\"\\nGENETIC ALGORITHM OPTIMIZATION COMPLETE\")\n", "print(f\"Function: {ga_results['objective_function']}\")\n", "print(f\"Dimensions: {ga_results['dimensions']}\")\n", "print(f\"Best fitness: {ga_results['best_fitness']:.6f}\")\n", "print(f\"Population size: {ga_results['population_size']}\")\n", "print(f\"Generations: {ga_results['generations']}\")\n", "\n", "final_stats = ga_results['final_population_stats']\n", "print(f\"\\nFinal population statistics:\")\n", "print(f\" Best: {final_stats['best_fitness']:.6f}\")\n", "print(f\" Average: {final_stats['average_fitness']:.6f}\")\n", "print(f\" Worst: {final_stats['worst_fitness']:.6f}\")\n", "print(f\" Std Dev: {final_stats['fitness_std']:.6f}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Example 2: Engineering Simulation\n", "\n", "Finite element analysis commonly run on SGE clusters:" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "@cluster(\n", " cores=12,\n", " memory=\"32GB\",\n", " time=\"04:00:00\",\n", " queue=\"all.q\",\n", " pe=\"mpi 12\" # MPI parallel environment\n", ")\n", "def finite_element_stress_analysis(mesh_density=\"medium\", material=\"steel\", load_cases=5):\n", " \"\"\"\n", " Simplified finite element stress analysis simulation.\n", " \"\"\"\n", " import numpy as np\n", " from scipy.sparse import csr_matrix\n", " from scipy.sparse.linalg import spsolve\n", " import math\n", " \n", " # Material properties\n", " materials = {\n", " 'steel': {'E': 200e9, 'nu': 0.3, 'yield_strength': 250e6, 'density': 7850},\n", " 'aluminum': {'E': 70e9, 'nu': 0.33, 'yield_strength': 276e6, 'density': 2700},\n", " 'titanium': {'E': 114e9, 'nu': 0.32, 'yield_strength': 880e6, 'density': 4500},\n", " 'concrete': {'E': 30e9, 'nu': 0.2, 'yield_strength': 30e6, 'density': 2400}\n", " }\n", " \n", " mat_props = materials.get(material, materials['steel'])\n", " E = mat_props['E'] # Young's modulus\n", " nu = mat_props['nu'] # Poisson's ratio\n", " yield_strength = mat_props['yield_strength']\n", " density = mat_props['density']\n", " \n", " print(f\"FEA Analysis - Material: {material}, Mesh: {mesh_density}, Load cases: {load_cases}\")\n", " \n", " # Mesh generation parameters\n", " mesh_sizes = {\n", " 'coarse': {'nx': 20, 'ny': 20, 'nz': 10},\n", " 'medium': {'nx': 40, 'ny': 40, 'nz': 20},\n", " 'fine': {'nx': 80, 'ny': 80, 'nz': 40}\n", " }\n", " \n", " mesh_params = mesh_sizes.get(mesh_density, mesh_sizes['medium'])\n", " nx, ny, nz = mesh_params['nx'], mesh_params['ny'], mesh_params['nz']\n", " \n", " # Geometry (simple beam)\n", " length, width, height = 2.0, 0.2, 0.1 # meters\n", " \n", " # Generate mesh\n", " def generate_3d_mesh(nx, ny, nz, length, width, height):\n", " \"\"\"Generate 3D hexahedral mesh\"\"\"\n", " nodes = []\n", " elements = []\n", " \n", " # Generate nodes\n", " for k in range(nz + 1):\n", " for j in range(ny + 1):\n", " for i in range(nx + 1):\n", " x = i * length / nx\n", " y = j * width / ny\n", " z = k * height / nz\n", " nodes.append([x, y, z])\n", " \n", " # Generate elements (hexahedral)\n", " for k in range(nz):\n", " for j in range(ny):\n", " for i in range(nx):\n", " # Node indices for hexahedral element\n", " n1 = k * (nx + 1) * (ny + 1) + j * (nx + 1) + i\n", " n2 = n1 + 1\n", " n3 = n1 + (nx + 1) + 1\n", " n4 = n1 + (nx + 1)\n", " n5 = n1 + (nx + 1) * (ny + 1)\n", " n6 = n5 + 1\n", " n7 = n5 + (nx + 1) + 1\n", " n8 = n5 + (nx + 1)\n", " \n", " elements.append([n1, n2, n3, n4, n5, n6, n7, n8])\n", " \n", " return np.array(nodes), np.array(elements)\n", " \n", " nodes, elements = generate_3d_mesh(nx, ny, nz, length, width, height)\n", " n_nodes = len(nodes)\n", " n_elements = len(elements)\n", " n_dof = n_nodes * 3 # 3 DOF per node (x, y, z displacements)\n", " \n", " print(f\"Mesh generated: {n_nodes:,} nodes, {n_elements:,} elements, {n_dof:,} DOF\")\n", " \n", " # Material matrix (isotropic elasticity)\n", " def material_matrix_3d(E, nu):\n", " \"\"\"3D elasticity matrix\"\"\"\n", " factor = E / ((1 + nu) * (1 - 2 * nu))\n", " D = np.zeros((6, 6))\n", " \n", " # Diagonal terms\n", " D[0, 0] = D[1, 1] = D[2, 2] = factor * (1 - nu)\n", " D[3, 3] = D[4, 4] = D[5, 5] = factor * (1 - 2 * nu) / 2\n", " \n", " # Off-diagonal terms\n", " D[0, 1] = D[0, 2] = D[1, 0] = D[1, 2] = D[2, 0] = D[2, 1] = factor * nu\n", " \n", " return D\n", " \n", " D_matrix = material_matrix_3d(E, nu)\n", " \n", " # Simplified stiffness matrix assembly\n", " def assemble_stiffness_matrix(nodes, elements, D_matrix):\n", " \"\"\"Assemble global stiffness matrix (simplified)\"\"\"\n", " K_global = np.zeros((n_dof, n_dof))\n", " \n", " for elem_idx, element in enumerate(elements[:min(1000, len(elements))]): # Limit for demo\n", " if elem_idx % 200 == 0:\n", " print(f\" Assembling element {elem_idx:,}/{len(elements):,}\")\n", " \n", " # Element nodes\n", " elem_nodes = nodes[element]\n", " \n", " # Simplified element stiffness (using average properties)\n", " volume = length * width * height / n_elements\n", " k_elem = volume * np.eye(24) * E / (length**2) # Simplified\n", " \n", " # Assembly\n", " for i, node_i in enumerate(element):\n", " for j, node_j in enumerate(element):\n", " for di in range(3):\n", " for dj in range(3):\n", " row = node_i * 3 + di\n", " col = node_j * 3 + dj\n", " if row < n_dof and col < n_dof:\n", " K_global[row, col] += k_elem[i*3+di, j*3+dj]\n", " \n", " return csr_matrix(K_global)\n", " \n", " print(\"Assembling stiffness matrix...\")\n", " K = assemble_stiffness_matrix(nodes, elements, D_matrix)\n", " \n", " # Load case analysis\n", " load_case_results = []\n", " \n", " for case in range(load_cases):\n", " print(f\"\\nAnalyzing load case {case + 1}/{load_cases}...\")\n", " \n", " # Define load case\n", " F = np.zeros(n_dof)\n", " \n", " if case == 0: # Point load at free end\n", " # Find nodes at free end (x = length)\n", " free_end_nodes = np.where(np.abs(nodes[:, 0] - length) < 1e-6)[0]\n", " if len(free_end_nodes) > 0:\n", " center_node = free_end_nodes[len(free_end_nodes)//2]\n", " F[center_node * 3 + 2] = -1000 # 1kN downward\n", " \n", " elif case == 1: # Distributed load\n", " # Apply distributed load to top surface\n", " top_nodes = np.where(np.abs(nodes[:, 2] - height) < 1e-6)[0]\n", " load_per_node = -100 # N per node\n", " for node in top_nodes:\n", " F[node * 3 + 2] = load_per_node\n", " \n", " elif case == 2: # Torsional load\n", " # Apply moments at free end\n", " free_end_nodes = np.where(np.abs(nodes[:, 0] - length) < 1e-6)[0]\n", " for node in free_end_nodes:\n", " y, z = nodes[node, 1], nodes[node, 2]\n", " # Simplified torsion as equivalent forces\n", " F[node * 3 + 1] = 500 * (z - height/2) # Simplified\n", " F[node * 3 + 2] = -500 * (y - width/2)\n", " \n", " elif case == 3: # Thermal expansion\n", " # Simplified thermal load (equivalent forces)\n", " alpha = 12e-6 # Thermal expansion coefficient\n", " delta_T = 100 # Temperature change (K)\n", " thermal_strain = alpha * delta_T\n", " # Apply as equivalent forces (simplified)\n", " F += np.random.normal(0, E * thermal_strain / 1000, n_dof)\n", " \n", " else: # Dynamic/random load\n", " # Random distributed forces\n", " np.random.seed(case * 123)\n", " F = np.random.normal(0, 50, n_dof)\n", " \n", " # Boundary conditions (fixed end)\n", " fixed_nodes = np.where(np.abs(nodes[:, 0]) < 1e-6)[0]\n", " fixed_dofs = []\n", " for node in fixed_nodes:\n", " fixed_dofs.extend([node * 3, node * 3 + 1, node * 3 + 2])\n", " \n", " # Apply boundary conditions\n", " K_reduced = K.copy()\n", " F_reduced = F.copy()\n", " \n", " # Zero out fixed DOFs\n", " for dof in fixed_dofs:\n", " if dof < n_dof:\n", " K_reduced[dof, :] = 0\n", " K_reduced[:, dof] = 0\n", " K_reduced[dof, dof] = 1\n", " F_reduced[dof] = 0\n", " \n", " # Solve for displacements\n", " print(\" Solving linear system...\")\n", " try:\n", " displacements = spsolve(K_reduced, F_reduced)\n", " except:\n", " # Fallback for singular matrices\n", " displacements = np.zeros(n_dof)\n", " print(\" Warning: Singular matrix, using zero displacements\")\n", " \n", " # Calculate stresses (simplified)\n", " max_displacement = np.max(np.abs(displacements))\n", " displacement_magnitude = np.sqrt(\n", " displacements[::3]**2 + displacements[1::3]**2 + displacements[2::3]**2\n", " )\n", " \n", " # Simplified stress calculation\n", " max_stress = E * max_displacement / length # Rough estimate\n", " \n", " # Safety factor\n", " safety_factor = yield_strength / max_stress if max_stress > 0 else float('inf')\n", " \n", " case_result = {\n", " 'case_id': case,\n", " 'load_type': ['point_load', 'distributed', 'torsion', 'thermal', 'dynamic'][case],\n", " 'max_displacement_m': max_displacement,\n", " 'max_stress_Pa': max_stress,\n", " 'safety_factor': min(safety_factor, 1000), # Cap at 1000\n", " 'total_force_N': np.sum(np.abs(F)),\n", " 'displacement_distribution': {\n", " 'mean': np.mean(displacement_magnitude),\n", " 'std': np.std(displacement_magnitude),\n", " 'max': np.max(displacement_magnitude)\n", " }\n", " }\n", " \n", " load_case_results.append(case_result)\n", " \n", " print(f\" Max displacement: {max_displacement:.2e} m\")\n", " print(f\" Max stress: {max_stress:.2e} Pa\")\n", " print(f\" Safety factor: {safety_factor:.2f}\")\n", " \n", " # Summary analysis\n", " max_displacement_overall = max(case['max_displacement_m'] for case in load_case_results)\n", " max_stress_overall = max(case['max_stress_Pa'] for case in load_case_results)\n", " min_safety_factor = min(case['safety_factor'] for case in load_case_results)\n", " \n", " analysis_results = {\n", " 'model_info': {\n", " 'material': material,\n", " 'mesh_density': mesh_density,\n", " 'nodes': n_nodes,\n", " 'elements': n_elements,\n", " 'dof': n_dof,\n", " 'geometry': {'length': length, 'width': width, 'height': height}\n", " },\n", " 'material_properties': mat_props,\n", " 'load_cases': load_case_results,\n", " 'summary': {\n", " 'max_displacement_m': max_displacement_overall,\n", " 'max_stress_Pa': max_stress_overall,\n", " 'min_safety_factor': min_safety_factor,\n", " 'critical_load_case': min(load_case_results, key=lambda x: x['safety_factor'])['load_type'],\n", " 'passes_safety_check': min_safety_factor > 2.0\n", " }\n", " }\n", " \n", " return analysis_results\n", "\n", "# Run FEA stress analysis\n", "fea_results = finite_element_stress_analysis(\n", " mesh_density=\"medium\", \n", " material=\"steel\", \n", " load_cases=3\n", ")\n", "\n", "print(f\"\\nFINITE ELEMENT ANALYSIS COMPLETE\")\n", "model_info = fea_results['model_info']\n", "print(f\"Material: {model_info['material']}\")\n", "print(f\"Mesh: {model_info['nodes']:,} nodes, {model_info['elements']:,} elements\")\n", "print(f\"DOF: {model_info['dof']:,}\")\n", "\n", "summary = fea_results['summary']\n", "print(f\"\\nSummary Results:\")\n", "print(f\" Max displacement: {summary['max_displacement_m']:.2e} m\")\n", "print(f\" Max stress: {summary['max_stress_Pa']:.2e} Pa\")\n", "print(f\" Min safety factor: {summary['min_safety_factor']:.2f}\")\n", "print(f\" Critical load case: {summary['critical_load_case']}\")\n", "print(f\" Passes safety check: {summary['passes_safety_check']}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Example 3: Multi-Objective Engineering Design\n", "\n", "Use SGE task arrays for design optimization:" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "@cluster(\n", " cores=6,\n", " memory=\"24GB\",\n", " time=\"02:00:00\",\n", " queue=\"all.q\",\n", " sge_array=\"1-25\" # SGE task array\n", ")\n", "def multi_objective_design_optimization(design_problem=\"beam_design\"):\n", " \"\"\"\n", " Multi-objective design optimization using SGE task arrays.\n", " Each task evaluates different design parameters.\n", " \"\"\"\n", " import os\n", " import numpy as np\n", " import random\n", " from math import pi, sqrt\n", " \n", " # Get SGE task array index\n", " task_id = int(os.environ.get('SGE_TASK_ID', '1'))\n", " \n", " print(f\"Design optimization task {task_id}\")\n", " \n", " def beam_design_objectives(width, height, length, material_density=7850):\n", " \"\"\"Calculate beam design objectives\"\"\"\n", " # Geometry constraints\n", " area = width * height\n", " moment_of_inertia = width * height**3 / 12\n", " volume = area * length\n", " mass = volume * material_density\n", " \n", " # Structural performance\n", " E = 200e9 # Young's modulus (Pa)\n", " max_load = 10000 # Maximum load (N)\n", " \n", " # Deflection calculation (simply supported beam)\n", " max_deflection = (5 * max_load * length**4) / (384 * E * moment_of_inertia)\n", " \n", " # Stress calculation\n", " max_moment = max_load * length / 4 # For simply supported beam\n", " max_stress = max_moment * (height / 2) / moment_of_inertia\n", " \n", " # Objectives to minimize\n", " objectives = {\n", " 'mass': mass, # Minimize weight\n", " 'deflection': max_deflection, # Minimize deflection\n", " 'stress': max_stress, # Minimize stress\n", " 'cost': mass * 2.5 + area * 10 # Material + manufacturing cost\n", " }\n", " \n", " # Constraints\n", " constraints = {\n", " 'deflection_limit': max_deflection < length / 250, # L/250 deflection limit\n", " 'stress_limit': max_stress < 250e6, # Yield stress limit\n", " 'aspect_ratio': height / width < 5, # Practical aspect ratio\n", " 'minimum_thickness': width > 0.01 and height > 0.01 # Minimum thickness\n", " }\n", " \n", " return objectives, constraints\n", " \n", " def truss_design_objectives(member_areas, topology, material_density=2700):\n", " \"\"\"Calculate truss design objectives\"\"\"\n", " # Simplified truss analysis\n", " n_members = len(member_areas)\n", " total_length = sum(topology) # Simplified total length\n", " total_volume = sum(area * length for area, length in zip(member_areas, topology))\n", " total_mass = total_volume * material_density\n", " \n", " # Simplified stiffness calculation\n", " E = 70e9 # Aluminum Young's modulus\n", " avg_stiffness = E * sum(member_areas) / n_members\n", " \n", " # Simplified stress analysis\n", " applied_load = 5000 # N\n", " avg_stress = applied_load / sum(member_areas)\n", " \n", " objectives = {\n", " 'mass': total_mass,\n", " 'compliance': 1 / avg_stiffness, # Inverse of stiffness\n", " 'max_stress': avg_stress,\n", " 'cost': total_mass * 3.0 + n_members * 50 # Material + connection cost\n", " }\n", " \n", " constraints = {\n", " 'stress_limit': avg_stress < 276e6, # Aluminum yield\n", " 'buckling_check': all(area > 1e-4 for area in member_areas), # Min area\n", " 'geometric_feasibility': len(member_areas) >= 3 # Minimum members\n", " }\n", " \n", " return objectives, constraints\n", " \n", " # Set up design space for this task\n", " np.random.seed(task_id * 42) # Reproducible but different per task\n", " \n", " if design_problem == \"beam_design\":\n", " # Generate design variables for beam\n", " width = np.random.uniform(0.05, 0.5) # 5cm to 50cm\n", " height = np.random.uniform(0.1, 1.0) # 10cm to 100cm\n", " length = np.random.uniform(2.0, 10.0) # 2m to 10m\n", " \n", " objectives, constraints = beam_design_objectives(width, height, length)\n", " design_vars = {'width': width, 'height': height, 'length': length}\n", " \n", " elif design_problem == \"truss_design\":\n", " # Generate design variables for truss\n", " n_members = random.randint(5, 15)\n", " member_areas = np.random.uniform(1e-4, 1e-2, n_members) # 1cm² to 100cm²\n", " topology = np.random.uniform(0.5, 3.0, n_members) # Member lengths\n", " \n", " objectives, constraints = truss_design_objectives(member_areas, topology)\n", " design_vars = {\n", " 'n_members': n_members,\n", " 'member_areas': member_areas.tolist(),\n", " 'topology': topology.tolist()\n", " }\n", " \n", " else:\n", " raise ValueError(f\"Unknown design problem: {design_problem}\")\n", " \n", " # Check constraint feasibility\n", " feasible = all(constraints.values())\n", " n_violated_constraints = sum(1 for satisfied in constraints.values() if not satisfied)\n", " \n", " # Calculate Pareto performance metrics\n", " def normalize_objectives(objectives):\n", " \"\"\"Normalize objectives for multi-objective comparison\"\"\"\n", " # Reference values for normalization (approximate)\n", " if design_problem == \"beam_design\":\n", " ref_values = {\n", " 'mass': 1000, # kg\n", " 'deflection': 0.01, # m\n", " 'stress': 100e6, # Pa\n", " 'cost': 5000 # currency units\n", " }\n", " else: # truss_design\n", " ref_values = {\n", " 'mass': 500, # kg\n", " 'compliance': 1e-9, # 1/N\n", " 'max_stress': 100e6, # Pa\n", " 'cost': 3000 # currency units\n", " }\n", " \n", " normalized = {}\n", " for obj, value in objectives.items():\n", " if obj in ref_values:\n", " normalized[obj] = value / ref_values[obj]\n", " else:\n", " normalized[obj] = value\n", " \n", " return normalized\n", " \n", " normalized_objectives = normalize_objectives(objectives)\n", " \n", " # Calculate aggregate performance metrics\n", " weighted_sum = sum(normalized_objectives.values()) # Equal weights\n", " max_objective = max(normalized_objectives.values())\n", " \n", " # Performance score (lower is better)\n", " if feasible:\n", " performance_score = weighted_sum\n", " else:\n", " # Penalty for infeasible designs\n", " performance_score = weighted_sum * (1 + 10 * n_violated_constraints)\n", " \n", " # Compile results\n", " design_result = {\n", " 'task_id': task_id,\n", " 'design_problem': design_problem,\n", " 'design_variables': design_vars,\n", " 'objectives': objectives,\n", " 'normalized_objectives': normalized_objectives,\n", " 'constraints': constraints,\n", " 'feasible': feasible,\n", " 'constraints_violated': n_violated_constraints,\n", " 'performance_metrics': {\n", " 'weighted_sum': weighted_sum,\n", " 'max_objective': max_objective,\n", " 'performance_score': performance_score\n", " },\n", " 'design_quality': {\n", " 'excellent': performance_score < 2.0 and feasible,\n", " 'good': performance_score < 4.0 and feasible,\n", " 'acceptable': performance_score < 8.0 and feasible,\n", " 'poor': not feasible or performance_score >= 8.0\n", " }\n", " }\n", " \n", " return design_result\n", "\n", "# Run design optimization (this would be one task of the SGE array)\n", "design_result = multi_objective_design_optimization(\"beam_design\")\n", "\n", "print(f\"\\nDESIGN OPTIMIZATION - Task {design_result['task_id']}\")\n", "print(f\"Problem: {design_result['design_problem']}\")\n", "print(f\"Feasible: {design_result['feasible']}\")\n", "\n", "if design_result['design_problem'] == 'beam_design':\n", " vars = design_result['design_variables']\n", " print(f\"\\nDesign Variables:\")\n", " print(f\" Width: {vars['width']:.3f} m\")\n", " print(f\" Height: {vars['height']:.3f} m\")\n", " print(f\" Length: {vars['length']:.3f} m\")\n", "\n", "print(f\"\\nObjectives:\")\n", "for obj, value in design_result['objectives'].items():\n", " if 'stress' in obj or 'deflection' in obj:\n", " print(f\" {obj}: {value:.2e}\")\n", " else:\n", " print(f\" {obj}: {value:.2f}\")\n", "\n", "perf = design_result['performance_metrics']\n", "print(f\"\\nPerformance Score: {perf['performance_score']:.2f}\")\n", "\n", "quality = design_result['design_quality']\n", "for level, is_level in quality.items():\n", " if is_level:\n", " print(f\"Design Quality: {level.upper()}\")\n", " break" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## SGE Parallel Environments and Resource Management" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def configure_sge_parallel_environments():\n", " \"\"\"\n", " Examples of different SGE parallel environment configurations.\n", " \"\"\"\n", " \n", " # Common SGE parallel environments\n", " pe_configs = {\n", " 'smp': {\n", " 'description': 'Symmetric Multi-Processing (shared memory)',\n", " 'use_case': 'Multi-threaded applications on single node',\n", " 'example_cores': [2, 4, 8, 16, 32],\n", " 'clustrix_config': {\n", " 'cores': 8,\n", " 'pe': 'smp 8',\n", " 'memory': '32GB'\n", " }\n", " },\n", " 'mpi': {\n", " 'description': 'Message Passing Interface (distributed memory)',\n", " 'use_case': 'Distributed parallel applications across nodes',\n", " 'example_cores': [8, 16, 32, 64, 128],\n", " 'clustrix_config': {\n", " 'cores': 32,\n", " 'pe': 'mpi 32',\n", " 'memory': '128GB'\n", " }\n", " },\n", " 'openmp': {\n", " 'description': 'OpenMP parallel environment',\n", " 'use_case': 'OpenMP applications with thread parallelism',\n", " 'example_cores': [4, 8, 12, 16],\n", " 'clustrix_config': {\n", " 'cores': 12,\n", " 'pe': 'openmp 12',\n", " 'memory': '48GB'\n", " }\n", " },\n", " 'hybrid': {\n", " 'description': 'Hybrid MPI+OpenMP',\n", " 'use_case': 'Applications using both MPI and OpenMP',\n", " 'example_cores': [16, 32, 64],\n", " 'clustrix_config': {\n", " 'cores': 32,\n", " 'pe': 'hybrid 32',\n", " 'memory': '128GB'\n", " }\n", " }\n", " }\n", " \n", " print(\"SGE Parallel Environment Configurations:\")\n", " print(\"=\" * 60)\n", " \n", " for pe_name, config in pe_configs.items():\n", " print(f\"\\n{pe_name.upper()}:\")\n", " print(f\" Description: {config['description']}\")\n", " print(f\" Use case: {config['use_case']}\")\n", " print(f\" Common core counts: {config['example_cores']}\")\n", " print(f\" Clustrix configuration:\")\n", " for key, value in config['clustrix_config'].items():\n", " print(f\" {key}: {value}\")\n", " \n", " return pe_configs\n", "\n", "# SGE resource selection helper\n", "def select_sge_resources(application_type, problem_size, parallelization=\"smp\"):\n", " \"\"\"\n", " Select appropriate SGE resources based on application characteristics.\n", " \"\"\"\n", " \n", " # Base resource requirements by application type\n", " app_requirements = {\n", " 'optimization': {'base_cores': 8, 'memory_per_core': 4, 'time_factor': 1.5},\n", " 'simulation': {'base_cores': 16, 'memory_per_core': 6, 'time_factor': 2.0},\n", " 'ml_training': {'base_cores': 4, 'memory_per_core': 8, 'time_factor': 1.0},\n", " 'data_analysis': {'base_cores': 6, 'memory_per_core': 4, 'time_factor': 0.8},\n", " 'engineering': {'base_cores': 12, 'memory_per_core': 5, 'time_factor': 1.8}\n", " }\n", " \n", " if application_type not in app_requirements:\n", " application_type = 'simulation' # Default\n", " \n", " req = app_requirements[application_type]\n", " \n", " # Scale resources based on problem size\n", " size_multipliers = {\n", " 'small': 0.5,\n", " 'medium': 1.0,\n", " 'large': 2.0,\n", " 'xlarge': 4.0\n", " }\n", " \n", " multiplier = size_multipliers.get(problem_size, 1.0)\n", " \n", " cores = max(1, int(req['base_cores'] * multiplier))\n", " memory_gb = max(4, int(cores * req['memory_per_core']))\n", " \n", " # Time estimation (hours)\n", " base_time = 2.0 # hours\n", " time_hours = max(0.5, base_time * req['time_factor'] * multiplier)\n", " \n", " # Format time as HH:MM:SS\n", " hours = int(time_hours)\n", " minutes = int((time_hours - hours) * 60)\n", " time_str = f\"{hours:02d}:{minutes:02d}:00\"\n", " \n", " # Queue selection\n", " if time_hours <= 1:\n", " queue = \"short.q\"\n", " elif time_hours <= 8:\n", " queue = \"all.q\"\n", " else:\n", " queue = \"long.q\"\n", " \n", " sge_config = {\n", " 'cores': cores,\n", " 'memory': f\"{memory_gb}GB\",\n", " 'time': time_str,\n", " 'queue': queue,\n", " 'pe': f\"{parallelization} {cores}\"\n", " }\n", " \n", " return sge_config\n", "\n", "# Display PE configurations\n", "pe_configs = configure_sge_parallel_environments()\n", "\n", "# Example resource selections\n", "print(\"\\n\\nSGE Resource Selection Examples:\")\n", "print(\"=\" * 60)\n", "\n", "examples = [\n", " ('optimization', 'medium', 'smp'),\n", " ('simulation', 'large', 'mpi'),\n", " ('ml_training', 'small', 'openmp'),\n", " ('engineering', 'xlarge', 'hybrid')\n", "]\n", "\n", "for app_type, size, parallel in examples:\n", " config = select_sge_resources(app_type, size, parallel)\n", " print(f\"\\n{app_type.upper()} ({size}, {parallel}):\")\n", " for key, value in config.items():\n", " print(f\" {key}: {value}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## SGE Job Monitoring and Management" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from clustrix import ClusterExecutor\n", "\n", "# Connect to SGE cluster and check status\n", "config = clustrix.get_config()\n", "executor = ClusterExecutor(config)\n", "\n", "try:\n", " executor.connect()\n", " print(\"✓ Successfully connected to SGE cluster\")\n", " \n", " # Check SGE version and configuration\n", " stdout, stderr = executor._execute_command(\"qconf -sconf\")\n", " if \"SGE\" in stdout or \"Grid Engine\" in stdout:\n", " print(\"✓ SGE/Grid Engine detected\")\n", " \n", " # List available queues\n", " stdout, stderr = executor._execute_command(\"qconf -sql\")\n", " if stdout:\n", " queues = stdout.strip().split('\\n')\n", " print(f\"\\nAvailable queues ({len(queues)}):\")\n", " for queue in queues[:10]: # Show first 10\n", " print(f\" {queue}\")\n", " if len(queues) > 10:\n", " print(f\" ... and {len(queues) - 10} more\")\n", " \n", " # List parallel environments\n", " stdout, stderr = executor._execute_command(\"qconf -spl\")\n", " if stdout:\n", " pes = stdout.strip().split('\\n')\n", " print(f\"\\nParallel environments ({len(pes)}):\")\n", " for pe in pes:\n", " print(f\" {pe}\")\n", " \n", " # Check queue status\n", " stdout, stderr = executor._execute_command(\"qstat -g c\")\n", " if stdout:\n", " print(\"\\nCluster queue summary:\")\n", " lines = stdout.strip().split('\\n')\n", " for line in lines[:15]: # Show header and first few lines\n", " print(f\" {line}\")\n", " \n", " # Check user's jobs\n", " username = config.username\n", " stdout, stderr = executor._execute_command(f\"qstat -u {username}\")\n", " if stdout and len(stdout.strip().split('\\n')) > 2:\n", " print(f\"\\nYour current jobs:\")\n", " lines = stdout.strip().split('\\n')\n", " for line in lines:\n", " print(f\" {line}\")\n", " else:\n", " print(f\"\\n✓ No jobs currently running for user {username}\")\n", " \n", " # Check host information\n", " stdout, stderr = executor._execute_command(\"qhost | head -20\")\n", " if stdout:\n", " print(\"\\nHost information (sample):\")\n", " lines = stdout.strip().split('\\n')\n", " for line in lines:\n", " print(f\" {line}\")\n", " \n", " executor.disconnect()\n", " print(\"\\n✓ SGE cluster monitoring completed successfully\")\n", " \n", "except Exception as e:\n", " print(f\"✗ Connection or monitoring failed: {e}\")\n", " print(\"Please check your SGE cluster configuration\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Summary\n", "\n", "This tutorial covered SGE cluster usage with Clustrix:\n", "\n", "1. **SGE Configuration** - Setting up Clustrix for SGE/Grid Engine clusters\n", "2. **Mathematical Optimization** - Genetic algorithms and complex optimization\n", "3. **Engineering Simulation** - Finite element analysis and structural design\n", "4. **Multi-Objective Design** - Engineering design optimization with task arrays\n", "5. **Parallel Environments** - SMP, MPI, OpenMP, and hybrid configurations\n", "6. **Resource Management** - Intelligent resource selection and queue management\n", "7. **Job Monitoring** - SGE cluster status and job management\n", "\n", "### Key SGE Features:\n", "\n", "- **Parallel Environments**: Use `pe` parameter for SMP, MPI, OpenMP configurations\n", "- **Task Arrays**: Efficient parameter sweeps with `sge_array` parameter\n", "- **Queue Selection**: Choose appropriate queues based on runtime requirements\n", "- **Resource Specification**: Flexible core, memory, and time allocation\n", "- **Job Dependencies**: Chain jobs with SGE dependency mechanisms\n", "- **Advanced Scheduling**: Priority, reservation, and resource policies\n", "\n", "### Best Practices:\n", "\n", "- **Parallel Environment Selection**: Choose PE based on application parallelization model\n", "- **Resource Estimation**: Use application profiling to estimate requirements accurately\n", "- **Queue Strategy**: Match job characteristics to appropriate queue policies\n", "- **Array Jobs**: Use task arrays for embarrassingly parallel workloads\n", "- **Monitoring**: Regular cluster status checks for optimal resource utilization\n", "\n", "### Next Steps:\n", "\n", "- Try [SLURM Tutorial](slurm_tutorial.ipynb) for SLURM-specific features\n", "- Explore [PBS Tutorial](pbs_tutorial.ipynb) for PBS/Torque clusters\n", "- Check [Kubernetes Tutorial](kubernetes_tutorial.ipynb) for containerized computing\n", "- Review [SSH Tutorial](ssh_tutorial.ipynb) for simple remote execution\n", "\n", "For more information, visit the [Clustrix Documentation](https://clustrix.readthedocs.io)." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.9.0" } }, "nbformat": 4, "nbformat_minor": 4 }